"""
闲鱼IM WebSocket客户端
参考: XianyuAutoAgent
"""
import asyncio
import base64
import json
import time
from datetime import datetime
from typing import Dict, Optional, Callable, Tuple
import websockets
from loguru import logger

from utils.xianyu_utils import generate_mid, generate_uuid, generate_device_id, decrypt


class XianyuIMClient:
    """闲鱼IM WebSocket客户端"""
    
    def __init__(self, cookies_dict: Dict[str, str], state_manager, browser_manager=None):
        self.base_url = 'wss://wss-goofish.dingtalk.com/'
        self.cookies_dict = cookies_dict
        self.state = state_manager
        self.browser = browser_manager  # ✅ 用于GoofishAPIClient的风控处理
        
        # 从cookies获取用户ID
        self.myid = cookies_dict.get('unb')
        if not self.myid:
            raise ValueError("未找到用户ID (unb)，无法连接IM")
        
        self.device_id = generate_device_id(self.myid)
        
        # WebSocket 相关
        self.ws = None
        self.connected = False
        self.running = False
        
        # 心跳配置
        self.heartbeat_interval = 15  # 15秒
        self.heartbeat_timeout = 5
        self.last_heartbeat_time = 0
        self.last_heartbeat_response = 0
        self.heartbeat_task = None
        
        # Token 配置
        self.token_refresh_interval = 3600  # 1小时
        self.last_token_refresh_time = 0
        self.current_token = None
        self.token_refresh_task = None
        
        # 消息处理回调
        self.message_callback = None
        
        # ✅ 会话创建回调：用于通知创建会话的结果
        self.conversation_created_callback = None  # (seller_id, item_id, chat_id) -> None
        
        # ✅ 存储待处理的会话创建请求：用于关联响应和请求
        # 格式：{mid: (seller_id, item_id, response_future)}
        self._pending_conversation_requests: Dict[str, tuple] = {}
        
        # ✅ 存储待处理的通用请求：用于请求-响应匹配
        # 格式：{mid: response_future}
        self._pending_requests: Dict[str, asyncio.Future] = {}
        
        # ✅ 存储 seller_id 到 chat_id 的映射
        self._chat_id_map: Dict[str, str] = {}
        
        # TCP客户端引用（用于上报消息）
        self.tcp_client = None
        
    async def get_token(self, device_id: str) -> Optional[str]:
        """获取IM Token - 使用GoofishAPIClient统一处理风控"""
        try:
            from modules.goofish_api import GoofishAPIClient, APIDefinition
            
            # ✅ 使用统一的API客户端（自动处理风控验证）
            api_client = GoofishAPIClient(self.state, self.browser)
            
            try:
                # 定义IM Token API
                im_token_api = APIDefinition(
                    api_name="mtop.taobao.idlemessage.pc.login.token",
                    version="1.0"
                )
                
                data = {
                    "appKey": "444e9908a51d1cb236a27862abc769c9",
                    "deviceId": device_id
                }
                
                # ✅ 使用统一的request方法（会自动处理风控、token刷新、session过期等）
                result = await api_client.request(im_token_api, data)
                
                # 检查返回码
                ret_codes = result.get("ret", [])
                if ret_codes and isinstance(ret_codes, list) and len(ret_codes) > 0:
                    ret_code = ret_codes[0]
                    # 判断是否成功
                    if isinstance(ret_code, str) and ret_code.startswith("SUCCESS"):
                        # 获取accessToken
                        if 'data' in result and 'accessToken' in result['data']:
                            self.current_token = result['data']['accessToken']
                            self.last_token_refresh_time = time.time()
                            logger.info("[IM] Token获取成功")
                            return self.current_token
                        else:
                            logger.warning(f"[IM] Token获取成功但缺少accessToken: {result}")
                            return None
                    else:
                        # ✅ 风控、封禁等错误已被GoofishAPIClient处理
                        logger.warning(f"[IM] Token获取失败，返回码: {ret_code}")
                        return None
                else:
                    logger.error(f"[IM] Token获取失败，无效的返回码: {result}")
                    return None
                    
            finally:
                # ✅ 确保关闭API客户端
                await api_client.close()
                        
        except Exception as e:
            logger.error(f"[IM] 获取Token异常: {e}")
            return None
    
    async def init_connection(self, ws):
        """初始化WebSocket连接"""
        # 获取token
        if not self.current_token:
            await self.get_token(self.device_id)
        
        if not self.current_token:
            raise Exception("[IM] 无法获取有效Token")
        
        # 发送注册消息
        reg_msg = {
            "lwp": "/reg",
            "headers": {
                "cache-header": "app-key token ua wv",
                "app-key": "444e9908a51d1cb236a27862abc769c9",
                "token": self.current_token,
                "ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36 DingTalk(2.1.5) OS(Windows/10) Browser(Chrome/133.0.0.0) DingWeb/2.1.5 IMPaaS DingWeb/2.1.5",
                "dt": "j",
                "wv": "im:3,au:3,sy:6",
                "sync": "0,0;0;0;",
                "did": self.device_id,
                "mid": generate_mid()
            }
        }
        
        # 注册和同步消息不输出日志（简化日志）
        await ws.send(json.dumps(reg_msg))
        await asyncio.sleep(1)
        
        # 发送同步状态消息
        sync_msg = {
            "lwp": "/r/SyncStatus/ackDiff",
            "headers": {"mid": generate_mid()},
            "body": [{
                "pipeline": "sync",
                "tooLong2Tag": "PNM,1",
                "channel": "sync",
                "topic": "sync",
                "highPts": 0,
                "pts": int(time.time() * 1000) * 1000,
                "seq": 0,
                "timestamp": int(time.time() * 1000)
            }]
        }
        
        await ws.send(json.dumps(sync_msg))
        logger.info('[IM] ✅ 连接注册完成，等待服务器响应...')
        self.connected = True
    
    async def send_message(self, chat_id: str, to_user_id: str, text: str, item_id: Optional[str] = None,
                          temp_id: Optional[str] = None, mid: Optional[str] = None):
        """
        发送消息给指定用户
        
        Args:
            chat_id: 会话ID（如果还没有会话，可以使用seller_id作为临时chat_id，服务器会自动创建会话）
            to_user_id: 接收者用户ID（纯数字seller_id）
            text: 消息内容
            item_id: 商品ID（可选，用于帮助服务器创建会话）
            temp_id: 前端生成的临时ID（用于乐观更新）
            mid: 消息mid（用于匹配官方响应，如果不提供则自动生成）
        
        Returns:
            (mid, msg_uuid, chat_id, to_user_id, text, item_id, message_id)
            message_id: 服务器返回的真实消息ID（格式为 "3802712875012.PNM"），如果服务器未返回则返回None
        """
        if not self.ws or not self.connected:
            raise Exception("[IM] WebSocket未连接")
        
        # ✅ 生成mid和uuid，用于后续错误关联
        # 如果外部提供了mid，则使用外部的（消息状态机要求）
        if not mid:
            mid = generate_mid()
        msg_uuid = generate_uuid()
        
        # ✅ 尝试从map中获取真实的chat_id（如果存在）
        # 优先使用 to_user_id (seller_id) 查找，避免使用临时的chat_id (即seller_id)
        if to_user_id in self._chat_id_map:
            real_chat_id = self._chat_id_map[to_user_id]
            # 只有当缓存的ID与传入的不同，或者传入的看起来像seller_id时才替换
            if real_chat_id != chat_id:
                logger.info(f"[IM] 纠正chat_id: {chat_id} -> {real_chat_id} (seller_id={to_user_id})")
                chat_id = real_chat_id
        
        # ✅ 确保cid格式正确（避免重复添加@goofish）
        cid = chat_id
        if "@goofish" not in cid:
            cid = f"{cid}@goofish"
            
        # ✅ 存储发送消息的请求，用于后续关联响应
        self._pending_message_requests = getattr(self, '_pending_message_requests', {})
        self._pending_message_requests[mid] = {
            'chat_id': chat_id,
            'to_user_id': to_user_id,
            'text': text,
            'item_id': item_id,
            'msg_uuid': msg_uuid,
            'message_type': 'text',  # ✅ 内容类型
            'custom_content_type': 1,  # ✅ 官方内容类型代码（1=文本）
            'content_type': 'private',  # ✅ 消息类型（私信）
            'temp_id': temp_id  # ✅ 存储前端临时ID
        }
        
        # 构建消息内容
        content = {
            "contentType": 1,
            "text": {
                "text": text
            }
        }
        content_base64 = base64.b64encode(json.dumps(content).encode('utf-8')).decode('utf-8')
        
        # ✅ 构建extJson（参考XianyuAutoAgent，首次发送时使用空对象）
        # 注意：根据参考项目，extJson在首次发送时是空的，服务器会根据actualReceivers和cid创建会话
        # 如果后续需要，可以尝试添加itemId，但先尝试空对象
        ext_json = "{}"
        
        # 构建消息
        msg = {
            "lwp": "/r/MessageSend/sendByReceiverScope",
            "headers": {
                "mid": mid
            },
            "body": [
                {
                    "uuid": msg_uuid,
                    "cid": cid,  # ✅ 使用已处理的cid
                    "conversationType": 1,  # 1表示单聊
                    "content": {
                        "contentType": 101,  # 自定义消息类型
                        "custom": {
                            "type": 1,  # 文本消息
                            "data": content_base64  # Base64编码的消息内容
                        }
                    },
                    "redPointPolicy": 0,
                    "extension": {
                        "extJson": ext_json  # ✅ 包含item_id以帮助创建会话
                    },
                    "ctx": {
                        "appVersion": "1.0",
                        "platform": "web"
                    },
                    "mtags": {},
                    "msgReadStatusSetting": 1
                },
                {
                    "actualReceivers": [
                        f"{to_user_id}@goofish",  # 接收者
                        f"{self.myid}@goofish"    # 发送者（自己）
                    ]
                }
            ]
        }
        
        # ✅ 只输出1对1单聊的消息内容（简化日志）
        await self.ws.send(json.dumps(msg))
        if item_id:
            logger.info(f"[IM] 📤 发送消息: {to_user_id} -> {text} (item_id={item_id})")
        else:
            logger.info(f"[IM] 📤 发送消息: {to_user_id} -> {text}")
        
        # ✅ 等待服务器响应（最多等待2秒），尝试获取真实的消息ID
        # 注意：服务器可能通过同步包返回消息，而不是直接响应
        # 所以这里只返回None，真实的message_id会在同步包中通过回调获取
        message_id = None
        
        # ✅ 返回mid、uuid和message_id（如果已获取）
        return mid, msg_uuid, chat_id, to_user_id, text, item_id, message_id
    
    async def send_image_message(self, chat_id: str, to_user_id: str, image_url: str, image_width: int = None, image_height: int = None, item_id: Optional[str] = None, temp_id: Optional[str] = None, mid: Optional[str] = None):
        """
        发送图片消息给指定用户
        
        Args:
            chat_id: 会话ID
            to_user_id: 接收者用户ID（纯数字seller_id）
            image_url: 图片URL（从上传API获取）
            image_width: 图片宽度（可选）
            image_height: 图片高度（可选）
            item_id: 商品ID（可选）
            temp_id: 前端生成的临时ID（用于乐观更新）
            mid: 消息ID（如果提供则使用，否则生成新的）
        
        Returns:
            (mid, msg_uuid, chat_id, to_user_id, image_url, item_id, message_id)
        """
        if not self.ws or not self.connected:
            raise Exception("[IM] WebSocket未连接")
        
        # ✅ 使用提供的mid或生成新的mid和uuid
        if not mid:
            mid = generate_mid()
        msg_uuid = generate_uuid()
        
        # ✅ 存储发送消息的请求，用于后续关联响应
        self._pending_message_requests = getattr(self, '_pending_message_requests', {})
        self._pending_message_requests[mid] = {
            'chat_id': chat_id,
            'to_user_id': to_user_id,
            'image_url': image_url,
            'image_width': image_width,
            'image_height': image_height,
            'item_id': item_id,
            'msg_uuid': msg_uuid,
            'temp_id': temp_id,  # ✅ 前端临时ID
            'message_type': 'image',  # ✅ 内容类型
            'custom_content_type': 2,  # ✅ 官方内容类型代码（2=图片）
            'content_type': 'private'  # ✅ 消息类型（私信）
        }
        
        # ✅ 构建图片消息内容（参考用户提供的格式）
        # content.custom.data 是base64编码的JSON，包含：
        # {"contentType":2,"image":{"pics":[{"height":153,"type":0,"url":"...","width":137}]}}
        image_data = {
            "contentType": 2,  # 图片类型
            "image": {
                "pics": [{
                    "height": image_height or 0,
                    "type": 0,
                    "url": image_url,
                    "width": image_width or 0
                }]
            }
        }
        image_data_base64 = base64.b64encode(json.dumps(image_data).encode('utf-8')).decode('utf-8')
        
        # ✅ 构建extJson
        ext_json = "{}"
        
        # 构建消息
        msg = {
            "lwp": "/r/MessageSend/sendByReceiverScope",
            "headers": {
                "mid": mid
            },
            "body": [
                {
                    "uuid": msg_uuid,
                    "cid": f"{chat_id}@goofish",
                    "conversationType": 1,  # 单聊
                    "content": {
                        "contentType": 101,  # 自定义消息类型
                        "custom": {
                            "type": 2,  # 图片消息
                            "data": image_data_base64
                        }
                    },
                    "redPointPolicy": 0,
                    "extension": {
                        "extJson": ext_json
                    },
                    "ctx": {
                        "appVersion": "1.0",
                        "platform": "web"
                    },
                    "mtags": {},
                    "msgReadStatusSetting": 1
                },
                {
                    "actualReceivers": [
                        f"{to_user_id}@goofish",
                        f"{self.myid}@goofish"
                    ]
                }
            ]
        }
        
        await self.ws.send(json.dumps(msg))
        logger.info(f"[IM] 📷 发送图片消息: {to_user_id} -> {image_url} (size={image_width}x{image_height})")
        
        # ✅ 返回mid、uuid和message_id（message_id会在同步包中获取）
        message_id = None
        return mid, msg_uuid, chat_id, to_user_id, image_url, item_id, message_id
    
    async def mark_messages_read(self, message_ids: list):
        """
        标记消息为已读（发送已读回执）
        
        Args:
            message_ids: 消息ID列表，格式为 ["3808247424509.PNM", ...]
        
        参考格式：
        {
            "lwp": "/r/MessageStatus/read",
            "headers": {"mid": "6001762364757416 0"},
            "body": [["3808247424509.PNM"]]
        }
        
        Returns:
            dict: 服务器响应，包含成功状态
            None: 发送失败或超时
        """
        if not self.ws or not self.connected:
            raise Exception("[IM] WebSocket未连接")
        
        if not message_ids:
            logger.warning("[IM] 标记已读失败：消息ID列表为空")
            return None
        
        try:
            # ✅ 构建已读回执消息
            # ⚠️ 注意：根据参考格式，body应该是二维数组：[[message_id1, message_id2, ...]]
            # 参考格式：{"lwp": "/r/MessageStatus/read", "headers": {"mid": "..."}, "body": [["3808247424509.PNM"]]}
            mid = generate_mid()
            
            # ✅ 创建Future用于等待响应
            response_future = asyncio.Future()
            self._pending_requests[mid] = response_future
            
            read_msg = {
                "lwp": "/r/MessageStatus/read",
                "headers": {
                    "mid": mid
                },
                "body": [message_ids]  # ✅ 消息ID数组包装在另一个数组中（二维数组格式）
            }
            
            logger.info(f"[IM] 准备发送已读回执: {len(message_ids)}条消息, mid={mid}, message_ids={message_ids[:5]}...")
            await self.ws.send(json.dumps(read_msg))
            
            # ✅ 等待服务器响应（最多5秒）
            try:
                response = await asyncio.wait_for(response_future, timeout=5.0)
                logger.info(f"[IM] ✅ 已读回执发送成功，服务器已确认: {len(message_ids)}条消息, mid={mid}")
                
                # ✅ 通知Backend已读回执发送成功
                if self.tcp_client:
                    await self.tcp_client.send({
                        'event': 'chat_read_confirmed',
                        'direction': 'res',
                        'data': {
                            'message_ids': message_ids,
                            'mid': mid,
                            'success': True
                        }
                    })
                
                return response
            except asyncio.TimeoutError:
                logger.error(f"[IM] ❌ 已读回执发送超时: {len(message_ids)}条消息, mid={mid}")
                # 清理超时的请求
                self._pending_requests.pop(mid, None)
                
                # ✅ 通知Backend已读回执发送失败
                if self.tcp_client:
                    await self.tcp_client.send({
                        'event': 'chat_read_confirmed',
                        'direction': 'res',
                        'data': {
                            'message_ids': message_ids,
                            'mid': mid,
                            'success': False,
                            'error': '超时'
                        }
                    })
                
                return None
        except Exception as e:
            logger.error(f"[IM] ❌ 发送已读回执异常: {e}")
            # 清理pending请求
            self._pending_requests.pop(mid, None)
            return None

    # todo:确认ws数据包格式
    async def recall_message(self, message_id: str, chat_id: str):
        """
        撤回消息
        
        Args:
            message_id: 要撤回的消息ID，格式为 "3808247424509.PNM"
            chat_id: 会话ID（chat_id），格式为 "55138335758" 或 "55138335758@goofish"
        
        参考格式（推测）：
        {
            "lwp": "/r/MessageRecall/recall",
            "headers": {"mid": "6001762364757416 0"},
            "body": [{
                "messageId": "3808247424509.PNM",
                "cid": "55138335758@goofish"
            }]
        }
        
        或者可能格式：
        {
            "lwp": "/r/Message/recall",
            "headers": {"mid": "6001762364757416 0"},
            "body": [{
                "messageId": "3808247424509.PNM",
                "cid": "55138335758@goofish"
            }]
        }
        """
        if not self.ws or not self.connected:
            raise Exception("[IM] WebSocket未连接")
        
        if not message_id:
            logger.warning("[IM] 撤回消息失败：消息ID为空")
            return False
        
        # ✅ 确保chat_id格式正确（添加@goofish后缀）
        if "@goofish" not in chat_id:
            chat_id = f"{chat_id}@goofish"
        
        # ✅ 构建撤回消息命令
        # 注意：撤回消息的命令格式可能需要根据实际日志调整
        mid = generate_mid()
        recall_msg = {
            "lwp": "/r/MessageRecall/recall",  # ✅ 撤回消息的lwp路径（可能需要根据实际调整）
            "headers": {
                "mid": mid
            },
            "body": [{
                "messageId": message_id,  # 要撤回的消息ID
                "cid": chat_id  # 会话ID
            }]
        }
        
        await self.ws.send(json.dumps(recall_msg))
        logger.info(f"[IM] ✅ 已发送撤回消息请求: message_id={message_id}, chat_id={chat_id}")
        
        # ✅ 等待服务器响应（最多2秒）
        # 注意：撤回消息可能需要等待服务器确认
        await asyncio.sleep(0.5)  # 短暂等待，确保消息已发送
        
        return True
    
    async def get_conversation_list(self, cursor: int = 9007199254740991, page_size: int = 50) -> Optional[dict]:
        """
        获取官方会话列表
        
        Args:
            cursor: 游标，首次请求使用 9007199254740991（最大值），后续使用返回的nextCursor
            page_size: 每页数量，默认50
        
        Returns:
            {
                "nextCursor": 下一页游标,
                "userConvs": 会话列表,
                "hasMore": 是否有更多
            }
        """
        if not self.ws or not self.connected:
            logger.error("[IM] WebSocket未连接，无法获取会话列表")
            return None
        
        try:
            # 生成唯一的mid
            mid = f"{int(datetime.now().timestamp() * 1000)} 0"
            
            # 创建Future用于接收响应
            response_future = asyncio.Future()
            self._pending_requests[mid] = response_future
            
            # 构建请求
            request = {
                "lwp": "/r/Conversation/listNewestPagination",
                "headers": {
                    "mid": mid
                },
                "body": [cursor, page_size]
            }
            
            # 发送请求
            await self.ws.send(json.dumps(request))
            logger.info(f"[IM] 📋 请求会话列表: cursor={cursor}, page_size={page_size}")
            
            # 等待响应（超时10秒）
            try:
                response = await asyncio.wait_for(response_future, timeout=10.0)
                
                if response.get("code") == 200:
                    body = response.get("body", {})
                    user_convs = body.get("userConvs", [])
                    logger.info(f"[IM] ✅ 获取会话列表成功: {len(user_convs)} 个会话")
                    return body
                else:
                    logger.error(f"[IM] ❌ 获取会话列表失败: code={response.get('code')}")
                    return None
            except asyncio.TimeoutError:
                logger.error("[IM] ❌ 获取会话列表超时")
                # 清理超时的请求
                self._pending_requests.pop(mid, None)
                return None
        except Exception as e:
            logger.error(f"[IM] ❌ 获取会话列表异常: {e}")
            return None
    
    async def create_conversation(self, seller_id: str, item_id: Optional[str] = None) -> Optional[str]:
        """
        创建单聊会话
        
        Args:
            seller_id: 卖家ID（纯数字）
            item_id: 商品ID（可选）
        
        Returns:
            Optional[str]: 返回创建的会话ID（chat_id），如果失败返回None
        """
        if not self.ws or not self.connected:
            raise Exception("[IM] WebSocket未连接")
        
        if not self.myid:
            logger.error("[IM] 无法创建会话：缺少用户ID (myid)")
            return None
        
        # ✅ 构建创建会话的消息（参考浏览器WebSocket消息）
        mid = generate_mid()
        create_msg = {
            "lwp": "/r/SingleChatConversation/create",
            "headers": {
                "mid": mid
            },
            "body": [{
                "pairFirst": f"{seller_id}@goofish",  # 卖家ID
                "pairSecond": f"{self.myid}@goofish",  # 我们的ID
                "bizType": "1",  # 业务类型：1表示单聊
                "extension": {
                    "itemId": item_id or "",  # 商品ID
                    "source": ""  # 来源
                },
                "ctx": {
                    "appVersion": "1.0",
                    "platform": "web"
                }
            }]
        }
        
        # ✅ 精简日志：创建会话不记录INFO级别日志
        
        # ✅ 存储请求信息，用于后续关联响应
        self._pending_conversation_requests[mid] = (seller_id, item_id)
        
        await self.ws.send(json.dumps(create_msg))
        
        # ✅ 等待服务器响应（通常很快，但需要等待）
        # 注意：服务器会返回会话信息，包括chat_id
        # 但由于是异步响应，我们需要在_handle_websocket_connection中处理
        # 这里等待一小段时间，让服务器处理
        await asyncio.sleep(0.5)
        
        # ✅ 注意：实际的chat_id会在服务器响应中返回
        # 我们需要在_handle_websocket_connection中处理响应并更新_chat_id_map
        # 这里暂时返回None，表示需要等待响应
        return None
    
    async def send_heartbeat(self):
        """发送心跳包"""
        if not self.ws:
            return
        
        heartbeat_msg = {
            "lwp": "/!",
            "headers": {
                "mid": generate_mid()
            }
        }
        await self.ws.send(json.dumps(heartbeat_msg))
        self.last_heartbeat_time = time.time()
        # 心跳包已发送（不记录日志，避免日志过多）
    
    async def heartbeat_loop(self):
        """心跳维护循环"""
        while self.running:
            try:
                current_time = time.time()
                
                # 检查是否需要发送心跳
                if current_time - self.last_heartbeat_time >= self.heartbeat_interval:
                    await self.send_heartbeat()
                
                # 检查心跳响应超时
                if (current_time - self.last_heartbeat_response) > (self.heartbeat_interval + self.heartbeat_timeout):
                    logger.warning("[IM] 心跳响应超时，连接可能已断开")
                    break
                
                await asyncio.sleep(1)
            except Exception as e:
                logger.error(f"[IM] 心跳循环出错: {e}")
                break
    
    def set_message_callback(self, callback):
        """设置消息回调函数"""
        self.message_callback = callback
    
    async def _handle_websocket_connection(self, websocket):
        """处理WebSocket连接的通用逻辑"""
        # 初始化连接
        await self.init_connection(websocket)
        
        # 初始化心跳时间
        self.last_heartbeat_time = time.time()
        self.last_heartbeat_response = time.time()
        
        # 启动心跳任务
        self.heartbeat_task = asyncio.create_task(self.heartbeat_loop())
        
        # ✅ 启动历史消息同步任务（已优化为批量处理，安全启用）
        asyncio.create_task(self._delayed_sync_history())
        
        # 接收消息循环
        async for message in websocket:
            try:
                message_data = json.loads(message)
                headers = message_data.get("headers", {})
                mid = headers.get("mid", "")
                
                # ✅ 优先处理通用请求响应（如获取会话列表）
                if mid and mid in self._pending_requests:
                    future = self._pending_requests.pop(mid)
                    if not future.done():
                        future.set_result(message_data)
                    continue  # 已处理，跳过后续逻辑
                
                # ✅ 处理会话创建响应（兼容旧格式：只有seller_id和item_id）
                if mid and mid in self._pending_conversation_requests:
                    request_data = self._pending_conversation_requests.pop(mid)
                    # ✅ 兼容两种格式：(seller_id, item_id) 或 (seller_id, item_id, future)
                    if len(request_data) == 2:
                        # 旧格式，只有seller_id和item_id，需要提取chat_id
                        seller_id, item_id = request_data
                        logger.debug(f"[IM] 收到会话创建响应（旧格式）: seller_id={seller_id}, item_id={item_id}")
                        
                        # ✅ 修复：提取chat_id并更新映射
                        body = message_data.get("body", {})
                        chat_id = (
                            body.get("cid") or 
                            body.get("conversationId") or 
                            body.get("chatId") or
                            body.get("singleChatConversation", {}).get("cid") or
                            body.get("conversation", {}).get("cid")
                        )
                        if chat_id:
                            # ✅ 使用与chat_manager_ws.py相同的键格式
                            key = (seller_id, item_id or '')
                            self._chat_id_map[key] = chat_id
                            logger.info(f"[IM] ✅ 会话创建成功: seller_id={seller_id}, item_id={item_id}, chat_id={chat_id}")
                        else:
                            logger.warning(f"[IM] ⚠️ 会话创建响应中未找到chat_id: seller_id={seller_id}, body={body}")
                    elif len(request_data) == 3:
                        # 新格式，包含Future
                        seller_id, item_id, response_future = request_data
                        if not response_future.done():
                            response_future.set_result(message_data)
                        logger.debug(f"[IM] 收到会话创建响应（新格式）: seller_id={seller_id}, item_id={item_id}")
                        
                        # ✅ 修复：新格式也需要提取chat_id
                        body = message_data.get("body", {})
                        chat_id = (
                            body.get("cid") or 
                            body.get("conversationId") or 
                            body.get("chatId") or
                            body.get("singleChatConversation", {}).get("cid") or
                            body.get("conversation", {}).get("cid")
                        )
                        if chat_id:
                            # ✅ 使用与chat_manager_ws.py相同的键格式
                            key = (seller_id, item_id or '')
                            self._chat_id_map[key] = chat_id
                            logger.info(f"[IM] ✅ 会话创建成功: seller_id={seller_id}, item_id={item_id}, chat_id={chat_id}")
                        else:
                            logger.warning(f"[IM] ⚠️ 会话创建响应中未找到chat_id: seller_id={seller_id}, body={body}")
                    continue  # 已处理，跳过后续逻辑
                
                # ✅ 兼容旧的会话创建响应处理（逐步废弃）
                is_conv_create_resp = self._is_conversation_create_response(message_data)
                if is_conv_create_resp:
                    await self._handle_conversation_create_response(message_data)
                    continue
                
                # ✅ 检查是否是发送消息的官方响应（通过mid匹配）
                if mid and hasattr(self, '_pending_message_requests') and mid in self._pending_message_requests:
                    # 这是发送消息的官方响应
                    await self._handle_send_message_response(message_data)
                    continue
                
                # ✅ 处理错误响应（code不为200的情况，且不是会话创建响应）
                if "code" in message_data and message_data.get("code") != 200:
                    error_code = message_data.get("code")
                    error_body = message_data.get("body", {})
                    error_reason = error_body.get("reason", "未知错误")
                    error_msg = error_body.get("developerMessage", error_reason)
                    
                    # ✅ 特殊处理"conversation not exist"错误
                    if "conversation not exist" in error_reason.lower() or "conversation not exist" in str(error_msg).lower():
                        logger.error(f"[IM] ❌ 会话不存在错误: {error_reason} (code={error_code})")
                        logger.error(f"[IM] 这通常意味着需要使用正确的chat_id或先创建会话")
                        # ✅ 不抛出异常，让上层处理（可能是首次发送消息，需要创建会话）
                    else:
                        logger.error(f"[IM] ❌ 服务器返回错误: code={error_code}, reason={error_reason}, msg={error_msg}")
                    
                    # 继续处理，不中断连接
                    continue
                
                # 处理心跳响应
                if self._is_heartbeat_response(message_data):
                    self.last_heartbeat_response = time.time()
                    # 收到心跳响应（不记录日志，避免日志过多）
                    continue
                
                # ✅ 调试：如果mid匹配但没被识别为会话创建响应，记录详细日志
                if mid and mid in self._pending_conversation_requests and not is_conv_create_resp:
                    logger.warning(f"[IM] ⚠️ 收到mid匹配的消息但未识别为会话创建响应:")
                    logger.warning(f"    mid={mid}, code={message_data.get('code', 'N/A')}, lwp={message_data.get('lwp', 'N/A')}")
                    logger.warning(f"    keys={list(message_data.keys())}")
                    logger.warning(f"    has_code={'code' in message_data}, has_body={'body' in message_data}")
                    logger.warning(f"    body_keys={list(message_data.get('body', {}).keys()) if isinstance(message_data.get('body'), dict) else 'N/A'}")
                    # ✅ 输出完整响应以便调试
                    logger.debug(f"    完整响应: {json.dumps(message_data, ensure_ascii=False)[:500]}")
                
                # 处理ACK响应（不记录日志，避免日志过多）
                # ✅ 注意：会话创建响应已经被处理了，这里不需要再检查
                if "headers" in message_data and "mid" in message_data["headers"]:
                    # ✅ 发送ACK响应（但跳过会话创建响应，因为它们已经在上面处理了）
                    ack = {
                        "code": 200,
                        "headers": {
                            "mid": message_data["headers"]["mid"],
                            "sid": message_data["headers"].get("sid", "")
                        }
                    }
                    await websocket.send(json.dumps(ack))
                    # ACK响应不记录日志
                
                # 处理同步包消息（聊天消息）
                await self._handle_sync_package(message_data)
                            
            except json.JSONDecodeError as e:
                logger.error(f"[IM] 消息解析失败: {e}, 原始消息: {message[:200]}")
            except Exception as e:
                logger.error(f"[IM] 处理消息时发生错误: {e}", exc_info=True)
    
    async def connect(self):
        """连接到闲鱼IM WebSocket"""
        self.running = True
        
        while self.running:
            try:
                # 构建Cookie字符串
                cookie_str = '; '.join([f"{k}={v}" for k, v in self.cookies_dict.items()])
                
                headers = {
                    "Cookie": cookie_str,
                    "Host": "wss-goofish.dingtalk.com",
                    "Connection": "Upgrade",
                    "Pragma": "no-cache",
                    "Cache-Control": "no-cache",
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36",
                    "Origin": "https://www.goofish.com",
                    "Accept-Encoding": "gzip, deflate, br, zstd",
                    "Accept-Language": "zh-CN,zh;q=0.9",
                }
                
                # 使用 additional_headers (websockets 11.0+)
                async with websockets.connect(
                    self.base_url,
                    additional_headers=headers,
                    ping_interval=None,  # 禁用默认ping，使用自定义心跳
                    ping_timeout=None
                ) as websocket:
                    self.ws = websocket
                    await self._handle_websocket_connection(websocket)
                            
            except websockets.exceptions.ConnectionClosed:
                logger.warning("[IM] WebSocket连接已关闭")
            except Exception as e:
                logger.error(f"[IM] 连接发生错误: {e}")
            finally:
                self.connected = False
                
                # 清理任务
                if self.heartbeat_task:
                    self.heartbeat_task.cancel()
                    try:
                        await self.heartbeat_task
                    except asyncio.CancelledError:
                        pass
                
                # 重连延迟
                if self.running:
                    logger.info("[IM] 等待5秒后重连...")
                    await asyncio.sleep(5)
                else:
                    break
    
    def _is_heartbeat_response(self, message_data: dict) -> bool:
        """判断是否为心跳响应"""
        try:
            return (
                isinstance(message_data, dict)
                and "headers" in message_data
                and "mid" in message_data["headers"]
                and "code" in message_data
                and message_data["code"] == 200
            )
        except Exception:
            return False
    
    def _is_sync_package(self, message_data: dict) -> bool:
        """判断是否为同步包消息"""
        try:
            return (
                isinstance(message_data, dict)
                and "body" in message_data
                and "syncPushPackage" in message_data["body"]
                and "data" in message_data["body"]["syncPushPackage"]
                and len(message_data["body"]["syncPushPackage"]["data"]) > 0
            )
        except Exception:
            return False
    
    def _is_chat_message(self, message: dict) -> bool:
        """判断是否为用户聊天消息"""
        try:
            return (
                isinstance(message, dict) 
                and "1" in message 
                and isinstance(message["1"], dict)
                and "10" in message["1"]
                and isinstance(message["1"]["10"], dict)
                and "reminderContent" in message["1"]["10"]
            )
        except Exception:
            return False
    
    def _is_conversation_create_response(self, message_data: dict) -> bool:
        """判断是否为会话创建响应"""
        # ✅ 优先通过mid匹配判断（最可靠的方式）
        headers = message_data.get("headers", {})
        mid = headers.get("mid", "")
        
        # ✅ 检查mid是否在待处理请求中
        if mid and mid in self._pending_conversation_requests:
            # 如果mid匹配待处理的会话创建请求，且响应包含code或body，说明是会话创建响应
            # 注意：响应可能没有lwp字段，所以不能依赖lwp判断
            # ✅ 只要有code或body字段，就认为是会话创建响应（成功或失败都算）
            has_code = "code" in message_data
            has_body = "body" in message_data
            if has_code or has_body:
                logger.debug(f"[IM] 🔍 mid匹配且包含code/body，确认为会话创建响应: mid={mid}, has_code={has_code}, has_body={has_body}")
                return True
            else:
                logger.debug(f"[IM] 🔍 mid匹配但缺少code/body字段: mid={mid}, keys={list(message_data.keys())}")
                return False  # ✅ 明确返回False，避免继续检查
        
        # ✅ 检查lwp是否为会话创建相关的响应（备用判断）
        lwp = message_data.get("lwp", "")
        if "/s/SingleChatConversation" in lwp or "/r/SingleChatConversation" in lwp:
            logger.debug(f"[IM] 🔍 lwp匹配，确认为会话创建响应: lwp={lwp}")
            return True
        
        # ✅ 或者检查body中是否包含会话信息（cid字段）
        if "body" in message_data and isinstance(message_data["body"], dict):
            body = message_data["body"]
            if "cid" in body or "conversationId" in body or "chatId" in body:
                logger.debug(f"[IM] 🔍 body包含cid/conversationId/chatId，确认为会话创建响应")
                return True
        
        return False
    
    async def _handle_conversation_create_response(self, message_data: dict):
        """处理会话创建响应"""
        try:
            # ✅ 获取请求的mid，用于关联待处理的请求
            headers = message_data.get("headers", {})
            mid = headers.get("mid", "")
            
            # ✅ 从待处理请求中获取seller_id和item_id
            seller_id = None
            item_id = None
            if mid and mid in self._pending_conversation_requests:
                seller_id, item_id = self._pending_conversation_requests.pop(mid)
                logger.debug(f"[IM] 关联会话创建请求: mid={mid}, seller_id={seller_id}, item_id={item_id}")
            else:
                logger.warning(f"[IM] ⚠️ 未找到会话创建请求的关联信息: mid={mid}")
            
            # ✅ 输出完整响应以便调试
            logger.debug(f"[IM] 会话创建响应完整内容: {json.dumps(message_data, ensure_ascii=False, indent=2)}")
            
            # ✅ 解析响应，提取chat_id
            # 实际响应格式（根据日志中的实际响应）：
            # {
            #   "headers": {"mid": "1011762366713205 0", ...},
            #   "code": 200,
            #   "body": {
            #     "singleChatConversation": {
            #       "cid": "55138335758@goofish",  # ✅ chat_id在这里
            #       ...
            #     },
            #     "lastMessage": {
            #       "message": {
            #         "cid": "55138335758@goofish",  # ✅ 也在这里
            #         ...
            #       }
            #     }
            #   }
            # }
            body = message_data.get("body", {})
            chat_id = None
            
            # ✅ 根据实际响应格式提取chat_id
            if isinstance(body, dict):
                # ✅ 优先从 singleChatConversation.cid 提取（这是实际响应中的字段）
                single_chat_conv = body.get("singleChatConversation", {})
                if isinstance(single_chat_conv, dict):
                    chat_id = single_chat_conv.get("cid")
                
                # ✅ 如果不存在，尝试从 lastMessage.message.cid 提取
                if not chat_id:
                    last_message = body.get("lastMessage", {})
                    if isinstance(last_message, dict):
                        message = last_message.get("message", {})
                        if isinstance(message, dict):
                            chat_id = message.get("cid")
                
                # ✅ 如果还是不存在，尝试直接从 body.cid 提取（旧格式）
                if not chat_id:
                    chat_id = body.get("cid")
                
                # ✅ 如果cid不存在，尝试其他可能的字段名
                if not chat_id:
                    chat_id = (body.get("conversationId") or 
                              body.get("chatId") or
                              body.get("conversation_id") or
                              body.get("chat_id"))
                
                if chat_id and isinstance(chat_id, str):
                    # 移除@goofish后缀（如果存在）
                    chat_id = chat_id.split('@')[0]
                    logger.debug(f"[IM] ✅ 成功提取chat_id: {chat_id}")
            elif isinstance(body, list) and len(body) > 0:
                # 如果body是数组，尝试从第一个元素提取
                first_item = body[0]
                if isinstance(first_item, dict):
                    chat_id = (first_item.get("cid") or
                              first_item.get("conversationId") or 
                              first_item.get("chatId") or
                              first_item.get("conversation_id") or
                              first_item.get("chat_id"))
                    if chat_id and isinstance(chat_id, str):
                        chat_id = chat_id.split('@')[0]
            
            # 如果从body中提取失败，检查是否有错误
            if not chat_id:
                # 检查是否有code字段，如果code!=200，说明创建失败
                code = message_data.get("code")
                if code and code != 200:
                    error_reason = message_data.get("body", {}).get("reason", "未知错误")
                    logger.error(f"[IM] ❌ 会话创建失败: code={code}, reason={error_reason}")
                else:
                    # ✅ code=200但未找到chat_id，可能是响应格式不同，输出完整响应以便调试
                    logger.warning(f"[IM] ⚠️ 会话创建响应中未找到chat_id（code={code}），完整响应: {json.dumps(message_data, ensure_ascii=False)[:1000]}")
            
            if chat_id:
                # ✅ 判断是新建还是已存在（通过createAt时间判断，如果创建时间很近可能是新建）
                create_at = body.get("createAt", 0) if isinstance(body, dict) else 0
                if create_at:
                    import time
                    create_time_ago = (time.time() * 1000 - create_at) / 1000  # 转换为秒
                    # ✅ 精简日志：会话创建成功不记录INFO级别日志
                    pass
                else:
                    # ✅ 精简日志：会话创建成功不记录INFO级别日志
                    pass
                
                # ✅ 通知回调（如果有）
                if self.conversation_created_callback and seller_id:
                    try:
                        await self.conversation_created_callback(seller_id, item_id, chat_id)
                        logger.debug(f"[IM] ✅ 会话创建回调已执行")
                    except Exception as e:
                        logger.error(f"[IM] 会话创建回调执行失败: {e}", exc_info=True)
                else:
                    logger.warning(f"[IM] ⚠️ 会话创建回调未设置或seller_id为空: callback={self.conversation_created_callback}, seller_id={seller_id}")
            else:
                logger.warning(f"[IM] ⚠️ 会话创建响应中未找到chat_id")
                
        except Exception as e:
            logger.error(f"[IM] 处理会话创建响应失败: {e}", exc_info=True)
            logger.debug(f"[IM] 原始响应: {json.dumps(message_data, ensure_ascii=False)[:1000]}")
    
    async def _handle_send_message_response(self, message_data: dict):
        """处理发送消息的官方响应（标准JSON格式）"""
        try:
            # 获取mid和响应码
            headers = message_data.get("headers", {})
            mid = headers.get("mid", "")
            code = message_data.get("code", 0)
            
            # 从待处理请求中获取请求信息
            request_info = self._pending_message_requests.pop(mid, None)
            if not request_info:
                logger.warning(f"[IM] ⚠️ 未找到发送消息请求的关联信息: mid={mid}")
                return
            
            # ✅ 修复：即使code != 200，也要尝试解析body，因为消息可能已经发送成功
            # 闲鱼IM响应有时会返回code=500，但消息实际已发送
            body = message_data.get("body", {})
            
            # 检查是否有有效的messageId（判断消息是否真正发送成功）
            message_id = body.get("messageId")
            
            if code != 200:
                # ✅ 输出完整响应，便于调试
                import json
                logger.error(f"[IM] 🔍 发送消息响应异常，完整数据: code={code}, mid={mid}, body={json.dumps(body, ensure_ascii=False)[:500]}")
                
                # ✅ 如果有messageId，说明消息实际发送成功了，只是响应码不标准
                if message_id:
                    logger.warning(f"[IM] ⚠️ 收到非标准响应码 code={code}，但消息已发送成功: messageId={message_id}, mid={mid}")
                else:
                    # 真正的失败：没有messageId且code != 200
                    logger.error(f"[IM] ❌ 发送消息失败: code={code}, mid={mid}, 无messageId")
                    
                    # ✅ 新增：上报失败给控制端
                    if self.message_callback:
                        error_payload = {
                            "client_id": self.state.get('client_id'),
                            "temp_id": request_info.get('temp_id'),
                            "mid": mid,
                            "success": False,
                            "error": f"发送失败(code={code}): {body.get('reason', '未知错误')}",
                            "message_id": None,
                            # ✅ 必须包含会话标识，否则前端无法找到对应会话
                            "seller_id": request_info.get('to_user_id'),
                            "item_id": request_info.get('item_id'),
                            "chat_id": request_info.get('chat_id')
                        }
                        try:
                            await self.message_callback(error_payload)
                        except Exception as e:
                            logger.error(f"[IM] 上报消息失败异常: {e}")
                            
                    return
            
            # 提取关键字段（官方API格式）
            message_id = body.get("messageId")  # 真实的消息ID
            # 提取时间戳
            create_at = body.get("createAt")
            
            # 提取内容
            content = body.get("content", {})
            custom = content.get("custom", {})
            message_text = custom.get("summary", "")
            
            # 提取扩展信息
            extension = body.get("extension", {})
            sender_user_id = extension.get("senderUserId", "")
            reminder_title = extension.get("reminderTitle", "")
            reminder_url = extension.get("reminderUrl", "")
            
            # ✅ 修复：如果sender_user_id为空，使用自己的ID（因为这是发送消息的响应，发送者肯定是自己）
            if not sender_user_id and self.myid:
                sender_user_id = self.myid
                logger.debug(f"[IM] ⚠️ 响应中缺少senderUserId，使用当前用户ID: {sender_user_id}")
            
            # 从reminderUrl提取item_id和chat_id
            # ⚠️ 注意：发送消息的官方响应中，reminderUrl的peerUserId是发送者自己，不是对方ID！
            import urllib.parse
            
            item_id = None
            chat_id = None
            
            if reminder_url:
                # 解析URL参数
                parsed = urllib.parse.urlparse(reminder_url)
                params = urllib.parse.parse_qs(parsed.query)
                # url_peer_user_id = params.get('peerUserId', [None])[0]  # ⚠️ 这是发送者自己，不需要
                item_id = params.get('itemId', [None])[0]
                chat_id = params.get('sid', [None])[0]
            
            # ✅ 对方ID应该从request_info中获取，不是从reminderUrl！
            # 因为发送消息的响应中，reminderUrl.peerUserId是发送者自己
            to_user_id = request_info.get('to_user_id')  # 这才是对方ID
            
            logger.info(f"[IM] ✅ 发送消息成功（官方响应）:")
            logger.info(f"    messageId: {message_id}")
            logger.info(f"    chat_id: {chat_id}")
            logger.info(f"    to_user_id: {to_user_id} (对方ID，从request_info)")
            logger.info(f"    sender_user_id: {sender_user_id} (我们自己)")
            logger.info(f"    content: {message_text}")
            
            # 推送到控制端（更新临时消息ID为真实ID）
            if self.message_callback and message_id and to_user_id:
                # ✅ 从request_info获取消息类型
                message_type = request_info.get('message_type', 'text')
                custom_content_type = request_info.get('custom_content_type', 1)
                content_type = request_info.get('content_type', 'private')
                
                # 构建消息数据（官方响应格式）
                # ✅ 关键：对方ID必须从request_info获取，不能从reminderUrl！
                # ⚠️ 注意：字段名必须与chat_manager_ws.py中的_on_im_message匹配！
                payload = {
                    "client_id": self.state.get('client_id'),
                    "sender_id": sender_user_id,  # ✅ 发送者ID（我们自己）← chat_manager_ws用sender_id
                    "sender_name": reminder_title if reminder_title else None,  # ← chat_manager_ws用sender_name
                    "to_user_id": to_user_id,  # ✅ 接收者ID（对方）- 用于官方响应时确定seller_id
                    "item_id": item_id or request_info.get('item_id'),
                    "chat_id": chat_id or request_info.get('chat_id'),
                    "message_id": message_id,  # ✅ 真实的消息ID（官方响应）
                    "message_type": message_type,  # ✅ 内容类型
                    "custom_content_type": custom_content_type,  # ✅ 官方内容类型代码
                    "content_type": content_type,  # ✅ 消息类型
                    "message": message_text or '[图片]' if message_type == 'image' else message_text,  # ← chat_manager_ws用message
                    "create_time": create_at,  # ← chat_manager_ws用create_time（毫秒）
                    # ========== 消息状态机：添加 temp_id 和 mid ==========
                    "temp_id": request_info.get('temp_id'),  # ✅ 前端临时ID
                    "mid": mid  # ✅ 消息mid（用于匹配）
                }
                
                # ✅ 如果是图片消息，添加图片信息
                if message_type == 'image':
                    payload["image_url"] = request_info.get('image_url')
                    payload["image_width"] = request_info.get('image_width')
                    payload["image_height"] = request_info.get('image_height')
                
                logger.debug(f"[IM] 📋 准备推送官方响应: sender_id={sender_user_id}, to_user_id={to_user_id}, message_id={message_id}, type={message_type}, content={message_text[:50] if message_text else payload.get('image_url', 'None')[:50]}")
                
                # 调用回调
                try:
                    await self.message_callback(payload)
                    logger.debug(f"[IM] 已推送发送消息到控制端（官方响应）: message_id={message_id}")
                except Exception as e:
                    logger.error(f"[IM] 消息回调执行失败: {e}", exc_info=True)
            
        except Exception as e:
            logger.error(f"[IM] 处理发送消息响应失败: {e}", exc_info=True)
            logger.debug(f"[IM] 原始响应: {json.dumps(message_data, ensure_ascii=False)[:1000]}")
    
    async def _handle_sync_package(self, message_data: dict):
        """处理同步包消息"""
        try:
            # ✅ 不再输出同步包消息的原始内容（日志太多且无实际价值）
            
            # 如果不是同步包消息，直接返回
            if not self._is_sync_package(message_data):
                return
            
            # 获取并解密数据
            sync_data = message_data["body"]["syncPushPackage"]["data"][0]
            
            if "data" not in sync_data:
                logger.debug("[IM] 同步包中无data字段")
                return
            
            # 解密数据
            data = sync_data["data"]
            decrypted_json = None
            try:
                # 尝试直接解析（未加密）
                decoded_bytes = base64.b64decode(data)
                decrypted_json = json.loads(decoded_bytes.decode("utf-8"))
                logger.debug("[IM] 消息未加密（直接Base64解码）")
            except Exception:
                # 需要解密（MessagePack）
                decrypted_str = decrypt(data)
                decrypted_json = json.loads(decrypted_str)
                logger.debug("[IM] 消息已解密（MessagePack）")
            
            # ✅ 提前检查是否是 sessionArouse 消息（"小闲鱼猜你想问"快捷回复），这类消息很长且无调试价值，直接跳过
            import json as json_module
            operation = decrypted_json.get("operation", {})
            content = operation.get("content", {}) if isinstance(operation, dict) else {}
            if isinstance(content, dict) and "sessionArouse" in content:
                logger.debug(f"[IM] ⏩ 跳过 sessionArouse 消息（快捷回复列表，无调试价值）")
                return
            
            # ✅ 打印完整的解密消息内容（DEBUG级别，便于调试）
            try:
                formatted_json = json_module.dumps(decrypted_json, indent=2, ensure_ascii=False)
                # ✅ 限制长度，避免超长输出
                if len(formatted_json) > 1000:
                    logger.debug(f"[IM] 📋 完整解密消息内容（JSON格式，截断前1000字符）:\n{formatted_json[:1000]}...")
                else:
                    logger.debug(f"[IM] 📋 完整解密消息内容（JSON格式）:\n{formatted_json}")
            except Exception as e:
                logger.debug(f"[IM] JSON格式化失败: {e}")
                logger.debug(f"[IM] 原始消息: {str(decrypted_json)[:1000]}")
            
            # ✅ 检查是否是撤回消息（可能在特定字段中）
            is_recall = False
            recall_message_id = None
            # 撤回消息可能没有reminderContent，或者结构不同
            # 检查多种可能的撤回消息特征
            if "1" in decrypted_json:
                msg_1 = decrypted_json["1"]
                # ✅ 添加类型检查：如果msg_1是字符串（消息ID），则跳过撤回消息检查
                if isinstance(msg_1, dict):
                    # 检查消息类型字段（可能在6字段中）
                    msg_6 = msg_1.get("6", {})
                    if isinstance(msg_6, dict):
                        msg_6_1 = msg_6.get("1")  # 消息类型代码
                        # 撤回消息可能有特定的类型代码
                    
                    # 检查10字段（reminder相关）
                    if "10" in msg_1:
                        msg_10 = msg_1["10"]
                        if isinstance(msg_10, dict):
                            # 检查撤回相关字段
                            reminder_content = msg_10.get("reminderContent", "")
                            reminder_notice = msg_10.get("reminderNotice", "")
                            # 撤回消息通常有特定的提示或字段
                            if "撤回" in str(reminder_content) or "撤回" in str(reminder_notice):
                                is_recall = True
                                # 尝试从extJson或其他字段提取被撤回的消息ID
                                ext_json_str = msg_10.get("extJson", "")
                                if ext_json_str:
                                    try:
                                        ext_json = json_module.loads(ext_json_str)
                                        if isinstance(ext_json, dict):
                                            recall_message_id = ext_json.get("recallMessageId") or ext_json.get("messageId")
                                    except Exception:
                                        pass
                            # ✅ 如果消息没有reminderContent但有其他撤回特征，也可能 is_recall
                            # 检查是否有recall相关的字段
                            if not is_recall and (msg_10.get("recallMessageId") or "recall" in str(msg_10).lower()):
                                is_recall = True
                                recall_message_id = msg_10.get("recallMessageId")
            
            # 判断是否为聊天消息
            is_chat_msg = self._is_chat_message(decrypted_json)
            
            # ✅ 如果是撤回消息，打印详细信息（INFO级别）
            if is_recall:
                logger.info(f"[IM] 🔄 收到撤回消息: recall_message_id={recall_message_id}, 完整结构:")
                try:
                    formatted_json = json_module.dumps(decrypted_json, indent=2, ensure_ascii=False)
                    logger.info(f"[IM] {formatted_json}")
                except Exception:
                    logger.info(f"[IM] {decrypted_json}")
                # 撤回消息也继续处理，以便更新前端
            
            # ✅ 如果既不是聊天消息也不是撤回消息，直接返回（不处理，已在上面输出了DEBUG日志）
            if not is_chat_msg and not is_recall:
                logger.debug(f"[IM] ⏩ 跳过非聊天消息且非撤回消息（如系统通知等）")
                return
            
            # ✅ 提取消息信息（添加类型检查，防止对字符串调用.get()）
            msg_1 = decrypted_json.get("1", {})
            if not isinstance(msg_1, dict):
                logger.warning(f"[IM] ⚠️ decrypted_json['1']不是字典类型: {type(msg_1)}, 值={msg_1}")
                logger.warning(f"[IM] 完整消息结构: {json_module.dumps(decrypted_json, indent=2, ensure_ascii=False)}")
                return
            
            msg_10 = msg_1.get("10", {})
            if not isinstance(msg_10, dict):
                logger.warning(f"[IM] ⚠️ decrypted_json['1']['10']不是字典类型: {type(msg_10)}, 值={msg_10}")
                logger.warning(f"[IM] 完整消息结构: {json_module.dumps(decrypted_json, indent=2, ensure_ascii=False)}")
                return
            
            # ✅ 安全提取create_time
            create_time = None
            if "5" in msg_1:
                create_time_val = msg_1["5"]
                if isinstance(create_time_val, (int, float, str)):
                    try:
                        create_time = int(create_time_val)
                    except (ValueError, TypeError):
                        logger.warning(f"[IM] ⚠️ 无法解析create_time: {create_time_val}")
                        return
                else:
                    logger.warning(f"[IM] ⚠️ create_time类型错误: {type(create_time_val)}")
                    return
            else:
                logger.warning(f"[IM] ⚠️ 消息中缺少create_time字段（'5'）")
                return
            
            # ✅ 确保msg_10是字典后再访问
            send_user_name = ""
            send_user_id = ""
            send_message = ""
            if isinstance(msg_10, dict):
                send_user_name = msg_10.get("reminderTitle", "")
                send_user_id = msg_10.get("senderUserId", "")
                send_message = msg_10.get("reminderContent", "")
                
                # ✅ 识别特殊系统消息：如果reminderTitle和reminderContent相同，说明这是系统消息，不应作为sender_name
                # 例如："未拍摄「开箱视频」"、"[我已修改价格，等待你付款]"等
                if send_user_name and send_message and send_user_name == send_message:
                    logger.info(f"[IM] 🔔 识别为系统消息（reminderTitle==reminderContent），清空sender_name: {send_user_name}")
                    send_user_name = ""  # ✅ 清空sender_name，避免覆盖真实的卖家昵称
            else:
                logger.warning(f"[IM] ⚠️ msg_10不是字典类型，无法提取消息内容: {type(msg_10)}, 值={msg_10}")
                return
            
            # ✅ 提取消息ID（格式为 "3802712875012.PNM"）- 先提取，后面会用到
            # 根据实际日志结构，消息ID在多个可能的路径下：
            # 1. decrypted_json["1"]["2"]["message"]["messageId"] (同步包消息)
            # 2. decrypted_json["body"]["lastMessage"]["message"]["messageId"] (会话创建响应)
            message_id = None
            
            # 路径1：从同步包消息中提取（decrypted_json["1"]["2"]["message"]["messageId"]）
            if "2" in msg_1:
                msg_2 = msg_1["2"]
                if isinstance(msg_2, dict):
                    message_obj = msg_2.get("message", {})
                    if isinstance(message_obj, dict):
                        message_id = message_obj.get("messageId")
            
            # 路径2：从body.lastMessage.message.messageId提取（会话创建响应）
            if not message_id and "body" in decrypted_json:
                body = decrypted_json["body"]
                if isinstance(body, dict):
                    last_msg = body.get("lastMessage", {})
                    if isinstance(last_msg, dict):
                        message_obj = last_msg.get("message", {})
                        if isinstance(message_obj, dict):
                            message_id = message_obj.get("messageId")
            
            # 路径3：直接从decrypted_json["1"]["3"]提取（消息ID在路径1.3下）
            if not message_id and "3" in msg_1:
                msg_3 = msg_1["3"]
                if isinstance(msg_3, str) and ".PNM" in msg_3:
                    # 消息ID在1.3路径下（格式如：3808547551687.PNM）
                    message_id = msg_3
            
            # 路径4：直接从decrypted_json["1"]["2"]提取（如果它是字符串格式的messageId）
            if not message_id and "2" in msg_1:
                msg_2 = msg_1["2"]
                if isinstance(msg_2, str) and ".PNM" in msg_2:
                    # 可能是直接的messageId字符串
                    message_id = msg_2
            
            # 如果仍然未找到，使用时间戳作为后备（但应该尽量避免）
            if not message_id:
                # ✅ 打印完整的解密消息内容，以便查找消息ID的位置
                import json as json_module
                logger.warning(f"[IM] ⚠️ 未找到消息ID，使用时间戳作为后备: {create_time}")
                logger.warning(f"[IM] 📋 完整解密消息内容（JSON格式）:")
                try:
                    # 格式化输出JSON，便于查看结构
                    formatted_json = json_module.dumps(decrypted_json, indent=2, ensure_ascii=False)
                    logger.warning(f"[IM] {formatted_json}")
                except Exception as e:
                    logger.warning(f"[IM] JSON格式化失败: {e}")
                    logger.warning(f"[IM] 原始消息: {decrypted_json}")
                
                # ✅ 尝试其他可能的路径查找消息ID
                logger.warning(f"[IM] 🔍 尝试查找消息ID的其他路径:")
                # 路径4：递归查找所有包含 "messageId" 或 ".PNM" 的字段
                def find_message_id_in_dict(d, path=""):
                    """递归查找消息ID"""
                    if isinstance(d, dict):
                        for key, value in d.items():
                            current_path = f"{path}.{key}" if path else key
                            if key == "messageId" or (isinstance(value, str) and ".PNM" in value):
                                logger.warning(f"[IM]   找到可能的消息ID: {current_path} = {value}")
                                return value
                            if isinstance(value, (dict, list)):
                                result = find_message_id_in_dict(value, current_path)
                                if result:
                                    return result
                    elif isinstance(d, list):
                        for i, item in enumerate(d):
                            current_path = f"{path}[{i}]"
                            result = find_message_id_in_dict(item, current_path)
                            if result:
                                return result
                    return None
                
                found_id = find_message_id_in_dict(decrypted_json)
                if found_id:
                    message_id = found_id
                    logger.warning(f"[IM] ✅ 通过递归查找找到消息ID: {message_id}")
                else:
                    message_id = f"{create_time}.PNM"
                    logger.warning(f"[IM] ❌ 未找到消息ID，使用时间戳: {message_id}")
            else:
                logger.debug(f"[IM] ✅ 成功提取消息ID: {message_id}")
            
            # 时效性验证（过滤5分钟前消息）
            message_age = time.time() * 1000 - create_time
            if message_age > 300000:  # 5分钟
                logger.debug(f"[IM] 过期消息，跳过（消息年龄: {message_age/1000:.1f}秒）")
                return
            
            # ✅ 获取商品ID和会话ID（添加类型检查）
            item_id = None
            # ✅ 从reminderUrl中提取真正的发送者ID（peerUserId）
            # 关键发现：接收到的消息中，senderUserId是接收者（我们自己）的ID
            # 真正的发送者ID在reminderUrl的peerUserId参数中
            actual_sender_id = None
            if isinstance(msg_10, dict):
                url_info = msg_10.get("reminderUrl", "")
                if url_info and isinstance(url_info, str):
                    if "itemId=" in url_info:
                        item_id = url_info.split("itemId=")[1].split("&")[0]
                    # ✅ 从reminderUrl中提取peerUserId（真正的发送者ID）
                    if "peerUserId=" in url_info:
                        actual_sender_id = url_info.split("peerUserId=")[1].split("&")[0]
                        logger.debug(f"[IM] 从reminderUrl提取发送者ID: peerUserId={actual_sender_id}")
            
            # ✅ chat_id从消息中提取，格式通常是 "chat_id@goofish"
            chat_id = None
            if isinstance(msg_1, dict) and "2" in msg_1:
                cid_str = msg_1["2"]
                if isinstance(cid_str, dict):
                    # 如果是字典，尝试从 message.cid 获取
                    message_obj = cid_str.get("message", {})
                    if isinstance(message_obj, dict):
                        cid = message_obj.get("cid", "")
                        if cid and isinstance(cid, str):
                            chat_id = cid.split('@')[0]
                elif isinstance(cid_str, str):
                    chat_id = cid_str.split('@')[0]
            
            # ✅ 判断消息方向：通过消息结构判断
            # 关键发现：接收到的消息中，senderUserId是接收者（我们自己）的ID，而不是发送者的ID
            # 真正的发送者ID需要通过chat_id映射反推，或者从消息结构的其他部分获取
            # 如果消息结构中有 msg_1["1"]["1"]["1"] 且等于 myid，说明这是别人发送给我们的消息（接收消息）
            # 如果消息结构简单（只有 message_id），可能是我们发送的消息的确认
            is_received_message = False
            actual_sender_id_from_structure = None
            
            # ✅ 方法1：从消息结构 msg_1["1"]["1"]["1"] 获取peer_user_id（接收者ID）
            # ⚠️ 关键：peer_user_id是接收者的ID
            # - 如果 peer_user_id == myid：这是接收的消息（对方发送给我们的）
            # - 如果 peer_user_id != myid：这是我们发送的消息（我们发送给对方的）
            peer_user_id = None
            if isinstance(msg_1, dict) and "1" in msg_1:
                msg_1_1 = msg_1["1"]
                if isinstance(msg_1_1, dict) and "1" in msg_1_1:
                    msg_1_1_1 = msg_1_1["1"]
                    if isinstance(msg_1_1_1, dict) and "1" in msg_1_1_1:
                        # 格式：msg_1["1"]["1"]["1"] = "670846218@goofish"
                        peer_user_id_str = msg_1_1_1.get("1", "")
                        if peer_user_id_str and isinstance(peer_user_id_str, str):
                            peer_user_id = peer_user_id_str.split('@')[0]
                            # ✅ 如果 peer_user_id 等于 myid，说明这是接收的消息（对方发送给我们的）
                            if self.myid and str(peer_user_id) == str(self.myid):
                                is_received_message = True
                                logger.debug(f"[IM] ✅ 识别为接收消息（通过消息结构）: peer_user_id={peer_user_id}, myid={self.myid}")
                                # ✅ 真正的发送者ID是actual_sender_id（从reminderUrl提取的peerUserId）
                                actual_sender_id_from_structure = actual_sender_id
                            else:
                                # ✅ 如果 peer_user_id 不等于 myid，说明这是我们发送的消息（我们发送给对方的）
                                logger.debug(f"[IM] ✅ 识别为发送消息（通过消息结构）: peer_user_id={peer_user_id}, myid={self.myid}")
                                # ✅ 这是我们发送的消息，应该跳过（因为发送的消息已经在send_message时推送了）
                                return
            
            # ✅ 方法2：如果actual_sender_id存在，使用它来判断（备用方法，如果方法1没有找到peer_user_id）
            # ⚠️ 注意：actual_sender_id是从reminderUrl提取的peerUserId，无论发送还是接收，都是对方ID
            # 所以不能直接用actual_sender_id判断消息方向，必须结合peer_user_id
            if not peer_user_id and actual_sender_id and self.myid:
                # ⚠️ 如果没有peer_user_id，无法准确判断消息方向
                # 这种情况下，如果actual_sender_id不等于myid，可能是接收消息
                # 但这不是100%准确的，应该优先使用方法1
                logger.warning(f"[IM] ⚠️ 无法从消息结构提取peer_user_id，使用actual_sender_id判断（可能不准确）: actual_sender_id={actual_sender_id}, myid={self.myid}")
                if str(actual_sender_id) != str(self.myid):
                    is_received_message = True
                    actual_sender_id_from_structure = actual_sender_id
                    logger.debug(f"[IM] ✅ 识别为接收消息（通过reminderUrl，备用方法）: actual_sender_id={actual_sender_id}, myid={self.myid}")
                else:
                    # 如果actual_sender_id等于myid，说明这是我们发送的消息
                    logger.debug(f"[IM] 跳过自己发送的消息（通过actual_sender_id，备用方法）: actual_sender_id={actual_sender_id}, myid={self.myid}")
                    return
            
            # ✅ 如果是自己发送的消息（不是接收消息），跳过（避免重复输出）
            if not is_received_message:
                # 简单结构的消息（只有message_id）通常是我们发送的消息的确认
                if isinstance(msg_1, dict) and len(msg_1) <= 6 and "1" in msg_1 and isinstance(msg_1["1"], str) and ".PNM" in msg_1["1"]:
                    logger.debug(f"[IM] 跳过自己发送的消息确认: message_id={msg_1.get('1')}, myid={self.myid}")
                    return
                # 如果senderUserId等于myid，但消息结构不是接收消息，也可能是我们发送的消息
                if send_user_id and self.myid and str(send_user_id) == str(self.myid):
                    logger.debug(f"[IM] 跳过自己发送的消息: sender_id={send_user_id}, myid={self.myid}")
                    return
                # ✅ 如果actual_sender_id等于myid，说明这是我们发送的消息
                if actual_sender_id and self.myid and str(actual_sender_id) == str(self.myid):
                    logger.debug(f"[IM] 跳过自己发送的消息（通过actual_sender_id）: actual_sender_id={actual_sender_id}, myid={self.myid}")
                    return
            
            # ✅ 如果是接收消息，但send_user_id是我们自己的ID，说明需要从其他地方获取真正的发送者ID
            # 真正的发送者ID将在chat_manager_ws中通过chat_id映射反推
            # 这里先保持send_user_id不变，让chat_manager_ws处理
            
            # ✅ 提取消息类型（文本、图片、语音等）
            message_type = "text"  # 默认内容类型
            custom_content_type = 1  # 默认官方内容类型代码（1=文本）
            content_type_category = "private"  # 默认消息类型（私信）
            image_url = None
            image_width = None
            image_height = None
            voice_url = None
            voice_duration = None
            voice_size_bytes = None
            system_data = None  # ✅ 系统消息数据（textCard）
            
            # ✅ 从消息内容中提取真实数据（在 "6" -> "3" -> "5" 字段中）
            # ✅ 使用已经检查过的msg_1，而不是直接访问decrypted_json["1"]
            msg_6 = msg_1.get("6", {}) if isinstance(msg_1, dict) else {}
            if isinstance(msg_6, dict):
                msg_6_3 = msg_6.get("3", {})
                if isinstance(msg_6_3, dict):
                    content_json_str = msg_6_3.get("5", "")  # JSON字符串，包含真实内容
                    content_type = msg_6_3.get("4")  # 内容类型：2=图片，3=语音
                    
                    if content_json_str:
                        try:
                            content_json = json_module.loads(content_json_str)
                            if isinstance(content_json, dict):
                                content_type_code = content_json.get("contentType")
                                
                                # ✅ 提取图片信息
                                if content_type_code == 2 or content_type == 2:
                                    message_type = "image"
                                    custom_content_type = 2  # 官方内容类型代码
                                    image_obj = content_json.get("image", {})
                                    if isinstance(image_obj, dict):
                                        pics = image_obj.get("pics", [])
                                        if pics and len(pics) > 0:
                                            pic = pics[0]  # 取第一张图片
                                            image_url = pic.get("url")
                                            image_width = pic.get("width")
                                            image_height = pic.get("height")
                                    logger.info(f"[IM] 📷 收到图片消息: message_id={message_id}, url={image_url}, size={image_width}x{image_height}")
                                    try:
                                        formatted_json = json_module.dumps(decrypted_json, indent=2, ensure_ascii=False)
                                        logger.info(f"[IM] {formatted_json}")
                                    except Exception:
                                        logger.info(f"[IM] {decrypted_json}")
                                
                                # ✅ 提取语音信息
                                elif content_type_code == 3 or content_type == 3:
                                    message_type = "voice"
                                    custom_content_type = 3  # 官方内容类型代码
                                    audio_obj = content_json.get("audio", {})
                                    if isinstance(audio_obj, dict):
                                        voice_url = audio_obj.get("url")
                                        voice_duration = audio_obj.get("duration")
                                        voice_size_bytes = audio_obj.get("sizeBytes")
                                    logger.info(f"[IM] 🎤 收到语音消息: message_id={message_id}, url={voice_url}, duration={voice_duration}s, size={voice_size_bytes}bytes")
                                
                                # ✅ 提取系统消息（textCard类型，contentType=6）
                                elif content_type_code == 6 or content_type == 6:
                                    message_type = "system"  # 标记为系统消息
                                    custom_content_type = 6  # 官方内容类型代码
                                    content_type_category = "system"  # 消息类型改为系统消息
                                    text_card = content_json.get("textCard", {})
                                    if isinstance(text_card, dict):
                                        # ✅ 提取textCard的完整信息，保存到message_extra中
                                        system_data = {
                                            "contentType": 6,
                                            "title": text_card.get("title", ""),
                                            "content": text_card.get("content", ""),
                                            "labelIcon": text_card.get("labelIcon", ""),
                                            "raw_textCard": text_card  # 保留完整的textCard数据
                                        }
                                        # ✅ 从textCard中提取可读的文本内容作为message_content
                                        system_title = text_card.get("title", "")
                                        system_content = text_card.get("content", "")
                                        # ✅ 如果reminderContent是标记格式，使用reminderNotice作为实际内容
                                        if isinstance(msg_10, dict):
                                            reminder_notice = msg_10.get("reminderNotice", "")
                                            if reminder_notice and reminder_notice != send_message:
                                                send_message = reminder_notice
                                        logger.info(f"[IM] 🔔 收到系统消息（textCard）: message_id={message_id}, title={system_title[:50]}")
                                
                        except Exception as e:
                            logger.debug(f"[IM] 解析消息内容JSON失败: {e}, content_json_str={content_json_str[:100]}")
            
            # ✅ 兼容旧逻辑：如果通过content_type没有识别到，使用文本匹配
            if message_type == "text":
                if "[图片]" in send_message or "[图" in send_message:
                    message_type = "image"
                    # 尝试从消息结构中提取图片URL
                    if isinstance(msg_10, dict):
                        reminder_url = msg_10.get("reminderUrl", "")
                        if reminder_url and ("image" in reminder_url.lower() or "img" in reminder_url.lower() or "pic" in reminder_url.lower()):
                            image_url = reminder_url
                            logger.info(f"[IM] 📷 从reminderUrl提取图片URL: {image_url}")
                        # 或者从extJson中提取
                        if not image_url:
                            ext_json_str = msg_10.get("extJson", "")
                            if ext_json_str:
                                try:
                                    ext_json = json_module.loads(ext_json_str)
                                    if isinstance(ext_json, dict):
                                        image_url = ext_json.get("imageUrl") or ext_json.get("picUrl") or ext_json.get("url") or ext_json.get("image_url")
                                        if image_url:
                                            logger.info(f"[IM] 📷 从extJson提取图片URL: {image_url}")
                                except Exception as e:
                                    logger.debug(f"[IM] 解析extJson失败: {e}")
                    # ✅ 如果还是找不到URL，尝试从整个消息结构中搜索
                    if not image_url:
                        # 尝试从decrypted_json的其他位置搜索图片URL
                        def find_image_url_in_dict(obj, depth=0):
                            """递归搜索字典中的图片URL"""
                            if depth > 5:  # 限制递归深度
                                return None
                            if isinstance(obj, dict):
                                for key, value in obj.items():
                                    if isinstance(key, str) and ("url" in key.lower() or "pic" in key.lower() or "image" in key.lower()):
                                        if isinstance(value, str) and ("http" in value or "https" in value):
                                            if any(x in value.lower() for x in ["image", "img", "pic", ".jpg", ".jpeg", ".png", ".gif", ".webp"]):
                                                return value
                                    if isinstance(value, (dict, list)):
                                        result = find_image_url_in_dict(value, depth + 1)
                                        if result:
                                            return result
                            elif isinstance(obj, list):
                                for item in obj:
                                    result = find_image_url_in_dict(item, depth + 1)
                                    if result:
                                        return result
                            return None
                        
                        found_url = find_image_url_in_dict(decrypted_json)
                        if found_url:
                            image_url = found_url
                            logger.info(f"[IM] 📷 从消息结构递归搜索找到图片URL: {image_url}")
                    
                    if image_url:
                        logger.info(f"[IM] 📷 收到图片消息（文本匹配）: message_id={message_id}, url={image_url}")
                    else:
                        logger.warning(f"[IM] ⚠️ 识别为图片消息但无法提取URL: message_id={message_id}, send_message={send_message[:50]}")
                        # ✅ 即使找不到URL，也保持为image类型，让后端知道这是图片消息
                elif "[语音]" in send_message or "[语音" in send_message:
                    message_type = "voice"
                    logger.info(f"[IM] 🎤 收到语音消息（文本匹配）: message_id={message_id}")
            
            if is_recall:
                message_type = "recall"
            else:
                # ✅ 普通文本消息，也输出完整结构（DEBUG级别）以便调试
                logger.debug(f"[IM] 📋 收到文本消息，完整结构:")
                try:
                    formatted_json = json_module.dumps(decrypted_json, indent=2, ensure_ascii=False)
                    logger.debug(f"[IM] {formatted_json}")
                except Exception:
                    logger.debug(f"[IM] {decrypted_json}")
            
            # ✅ 只输出1对1单聊的消息内容（简化日志）
            logger.info(f"[IM] 📥 收到消息: {send_user_name} -> {send_message} (message_id={message_id}, type={message_type})")
            
            # ✅ 提取回复消息ID（如果存在）
            reply_message_id = None
            if isinstance(msg_10, dict):
                # 路径1：直接从replyMessageId字段提取
                reply_message_id = msg_10.get("replyMessageId")
                # 路径2：如果不存在，尝试从extJson JSON字符串中解析
                if not reply_message_id:
                    ext_json_str = msg_10.get("extJson", "")
                    if ext_json_str and isinstance(ext_json_str, str):
                        try:
                            ext_json = json_module.loads(ext_json_str)
                            if isinstance(ext_json, dict):
                                reply_message_id = ext_json.get("replyMessageId")
                        except Exception:
                            pass
            
            # 调用回调函数
            if self.message_callback:
                # ✅ 重要：如果actual_sender_id存在，使用actual_sender_id作为sender_id（这是真正的发送者ID）
                # 如果actual_sender_id不存在，使用send_user_id（但send_user_id可能是接收者的ID，需要chat_manager_ws处理）
                final_sender_id = actual_sender_id_from_structure if actual_sender_id_from_structure else send_user_id
                
                callback_data = {
                    "chat_id": chat_id,
                    "sender_id": final_sender_id,  # ✅ 使用真正的发送者ID
                    "sender_name": send_user_name,
                    "item_id": item_id,
                    "message": send_message,
                    "create_time": create_time,
                    "message_id": message_id,  # ✅ 传递真实的消息ID
                    "reply_message_id": reply_message_id,  # ✅ 传递回复的消息ID（如果存在）
                    "message_type": message_type,  # ✅ 内容类型（text/image/voice/recall）
                    "custom_content_type": custom_content_type,  # ✅ 官方内容类型代码
                    "content_type": content_type_category,  # ✅ 消息类型（private/system）
                    "raw_data": decrypted_json  # ✅ 传递原始消息数据，便于后续处理
                }
                
                # ✅ 如果是图片消息，添加图片信息
                if message_type == "image" and image_url:
                    callback_data["image_url"] = image_url
                    if image_width:
                        callback_data["image_width"] = image_width
                    if image_height:
                        callback_data["image_height"] = image_height
                
                # ✅ 如果是语音消息，添加语音信息
                if message_type == "voice" and voice_url:
                    callback_data["voice_url"] = voice_url
                    if voice_duration is not None:
                        callback_data["voice_duration"] = voice_duration
                    if voice_size_bytes is not None:
                        callback_data["voice_size_bytes"] = voice_size_bytes
                
                # ✅ 如果是撤回消息，添加撤回相关信息
                if is_recall and recall_message_id:
                    callback_data["recall_message_id"] = recall_message_id
                    callback_data["is_recall"] = True
                
                # ✅ 如果是系统消息（textCard），添加系统消息数据
                if message_type == "system" and system_data:
                    callback_data["system_data"] = system_data
                    # ✅ 使用reminderNotice作为消息内容（更完整的文本）
                    if isinstance(msg_10, dict):
                        reminder_notice = msg_10.get("reminderNotice", "")
                        if reminder_notice:
                            callback_data["message"] = reminder_notice
                
                await self.message_callback(callback_data)
                
        except Exception as e:
            logger.error(f"[IM] 处理同步包消息失败: {e}", exc_info=True)
            logger.debug(f"[IM] 原始消息: {json.dumps(message_data, ensure_ascii=False, indent=2)[:1000]}")
    
    async def disconnect(self):
        """断开连接"""
        self.running = False
        if self.ws:
            await self.ws.close()
    
    async def sync_conversations_to_backend(self):
        """同步会话列表到后端"""
        try:
            # 获取会话列表
            result = await self.get_conversation_list()
            if not result:
                logger.warning("[IM] 获取会话列表失败，跳过同步")
                return
            
            user_convs = result.get("userConvs", [])
            logger.info(f"[IM] 📋 开始同步 {len(user_convs)} 个会话")
            
            # 解析并上报到后端
            synced_count = 0
            for conv_data in user_convs:
                if await self._parse_and_report_conversation(conv_data):
                    synced_count += 1
            
            logger.info(f"[IM] ✅ 会话同步完成: 成功{synced_count}/{len(user_convs)}")
        except Exception as e:
            logger.error(f"[IM] ❌ 同步会话失败: {e}")
    
    async def _parse_and_report_conversation(self, conv_data: dict) -> bool:
        """解析并上报单个会话"""
        try:
            # 只处理单聊
            if conv_data.get("type") != 1:
                return False
            
            single_chat = conv_data.get("singleChatUserConversation", {})
            conversation = single_chat.get("singleChatConversation", {})
            last_msg = single_chat.get("lastMessage", {})
            
            # 提取基本信息
            cid = conversation.get("cid", "").replace("@goofish", "")
            extension = conversation.get("extension", {})
            
            # ✅ 修复seller_id提取逻辑：优先使用itemSellerId（商品卖家）
            # 因为账号可能是买家账号，extUserId可能是买家自己
            seller_id = extension.get("itemSellerId") or extension.get("ownerUserId") or extension.get("extUserId")
            item_id = extension.get("itemId")
            item_title = extension.get("itemTitle")
            item_main_pic = extension.get("itemMainPic")
            
            # ✅ 提取卖家名称（从对话扩展信息中）
            seller_name = extension.get("extUserNick") or extension.get("title") or ""
            
            # ✅ 如果没有卖家名称，尝试从sender中获取
            if not seller_name:
                # 获取对方用户信息
                target = single_chat.get("target", {})
                if target:
                    seller_name = target.get("userNick") or target.get("nick") or ""
            
            if not seller_id or not item_id:
                logger.debug(f"[IM] 跳过会话（缺少seller_id或item_id）: cid={cid}")
                return False
            
            # 提取最后一条消息
            if not last_msg:
                logger.debug(f"[IM] 跳过会话（无最后消息）: cid={cid}, seller_id={seller_id}")
                return False
            
            message_data = last_msg.get("message", {})
            message_id = message_data.get("messageId")
            content = message_data.get("content", {})
            sender = message_data.get("sender", {})
            create_at = message_data.get("createAt")
            
            # 判断消息方向
            sender_uid = sender.get("uid", "").replace("@goofish", "")
            # 需要myid来判断方向
            direction = "send" if sender_uid == str(self.myid) else "recv"
            
            # 解析消息内容和类型
            message_content, message_type = self._parse_message_content_and_type(content)
            
            # ✅ 保护逻辑：如果seller_name与message_content相同，说明这是系统消息，清空seller_name
            # 例如："未拍摄「开箱视频」"、"[我已修改价格，等待你付款]"、"交易消息"等
            if seller_name and message_content and seller_name == message_content:
                logger.info(f"[IM] 🔔 识别为系统消息（seller_name==message_content），清空seller_name: {seller_name}")
                seller_name = ""
            
            # 构建上报数据
            report_data = {
                "event": "chat_message",
                "direction": "req",
                "data": {
                    "chat_id": cid,  # ← 使用官方的conversation_id
                    "seller_id": seller_id,
                    "seller_name": seller_name,  # ✅ 添加卖家名称
                    "item_id": item_id,
                    "item_title": item_title,
                    "message_id": message_id,
                    "message_type": message_type,
                    "direction": direction,  # ← 关键！
                    "message_content": message_content,
                    "message_time": self._parse_timestamp(create_at)
                }
            }
            
            # 如果是图片消息，添加图片信息
            if message_type == "image":
                image_info = self._extract_image_info(content)
                if image_info:
                    report_data["data"]["image_url"] = image_info.get("url")
                    report_data["data"]["image_width"] = image_info.get("width")
                    report_data["data"]["image_height"] = image_info.get("height")
            
            # 上报到后端（通过TCP客户端）
            if self.tcp_client:
                await self.tcp_client.send(report_data)
                logger.info(f"[IM] ✅ 已上报会话: cid={cid}, seller_id={seller_id}, seller_name={seller_name}, direction={direction}, event={report_data.get('event')}, req_direction={report_data.get('direction')}")
                return True
            else:
                logger.warning("[IM] TCP客户端未初始化，无法上报会话")
                return False
        except Exception as e:
            logger.error(f"[IM] ❌ 解析会话失败: {e}")
            return False
    
    def _parse_message_content_and_type(self, content: dict) -> tuple:
        """解析消息内容和类型"""
        import base64
        import json as json_module
        
        if content.get("contentType") == 101:  # 自定义消息
            custom = content.get("custom", {})
            summary = custom.get("summary", "")
            msg_type = custom.get("type", 1)
            
            if msg_type == 2:
                return summary or "[图片]", "image"
            elif msg_type == 3:
                return summary or "[语音]", "voice"
            else:
                # 尝试解析data字段
                try:
                    data = base64.b64decode(custom.get("data", "")).decode("utf-8")
                    data_json = json_module.loads(data)
                    
                    # 提取文本内容
                    text = data_json.get("text", {}).get("text", "")
                    return text or summary, "text"
                except:
                    return summary, "text"
        
        # 默认文本消息
        return content.get("text", "[消息]"), "text"
    
    def _extract_image_info(self, content: dict) -> Optional[dict]:
        """提取图片信息"""
        import base64
        import json as json_module
        
        try:
            if content.get("contentType") == 101:
                custom = content.get("custom", {})
                if custom.get("type") == 2:  # 图片消息
                    data = base64.b64decode(custom.get("data", "")).decode("utf-8")
                    data_json = json_module.loads(data)
                    
                    image_obj = data_json.get("image", {})
                    pics = image_obj.get("pics", [])
                    if pics and len(pics) > 0:
                        pic = pics[0]
                        return {
                            "url": pic.get("url"),
                            "width": pic.get("width"),
                            "height": pic.get("height")
                        }
        except Exception as e:
            logger.debug(f"[IM] 提取图片信息失败: {e}")
        
        return None
    
    def _parse_timestamp(self, timestamp: int) -> str:
        """解析时间戳（毫秒）"""
        if not timestamp:
            return datetime.now().isoformat()
        
        try:
            dt = datetime.fromtimestamp(timestamp / 1000)
            return dt.isoformat()
        except:
            return datetime.now().isoformat()
    
    # ========== 历史消息同步功能 ==========
    
    async def _delayed_sync_history(self):
        """延迟启动历史消息同步（等待连接稳定）"""
        try:
            # 等待5秒确保连接稳定
            await asyncio.sleep(5)
            
            logger.info("[IM] ⏰ 准备开始历史数据同步...")
            
            # ✅ 步骤1：先同步会话列表，确保会话记录存在
            logger.info("[IM] 📋 步骤1/2：同步会话列表到后端...")
            await self.sync_conversations_to_backend()
            
            # 等待1秒，让会话数据保存完成
            await asyncio.sleep(1)
            
            # ✅ 步骤2：再同步历史消息
            logger.info("[IM] 📥 步骤2/2：开始同步历史消息...")
            # ✅ 减少每个会话同步的消息数（从50减到30），避免数据库连接池耗尽
            await self.sync_all_conversations_history(limit_per_conv=30)
            
            logger.info("[IM] ✅ 历史数据同步完成！")
            
        except Exception as e:
            logger.error(f"[IM] ❌ 延迟同步历史数据失败: {e}", exc_info=True)
    
    async def list_user_messages(self, chat_id: str, cursor: int = 9007199254740991, page_size: int = 20) -> Optional[dict]:
        """
        获取会话的历史消息列表
        
        Args:
            chat_id: 会话ID
            cursor: 游标，首次请求使用9007199254740991，后续使用响应中的nextCursor
            page_size: 每页数量，默认20
        
        Returns:
            响应数据，包含消息列表和分页信息
        """
        if not self.ws or not self.connected:
            logger.warning("[IM] WebSocket未连接，无法获取历史消息")
            return None
        
        try:
            mid = generate_mid()
            
            # 构建请求
            request_data = {
                "lwp": "/r/MessageManager/listUserMessages",
                "headers": {"mid": mid},
                "body": [
                    f"{chat_id}@goofish",  # 完整的会话ID
                    False,  # 是否只获取未读
                    cursor,  # 游标
                    page_size,  # 每页数量
                    False  # 是否倒序
                ]
            }
            
            logger.info(f"[IM] 📤 准备发送请求: mid={mid}, chat_id={chat_id}@goofish")
            logger.info(f"[IM] 📤 请求内容: {request_data}")
            
            # 创建Future用于等待响应
            response_future = asyncio.Future()
            self._pending_requests[mid] = response_future
            
            # 发送请求
            await self.ws.send(json.dumps(request_data))
            logger.info(f"[IM] ✅ 已发送历史消息请求: chat_id={chat_id}, cursor={cursor}, page_size={page_size}, mid={mid}")
            
            # 等待响应（超时10秒）
            try:
                logger.info(f"[IM] ⏳ 等待响应: mid={mid}")
                response = await asyncio.wait_for(response_future, timeout=10.0)
                
                logger.info(f"[IM] 📥 收到响应: mid={mid}, response keys={list(response.keys()) if response else None}")
                
                # 检查响应码
                code = response.get('code')
                logger.info(f"[IM] 📥 响应码: {code}")
                
                if code == 200:
                    body = response.get('body', {})
                    has_more = body.get('hasMore', False)
                    next_cursor = body.get('nextCursor')
                    messages = body.get('userMessageModels', [])
                    logger.info(f"[IM] ✅ 获取历史消息成功: chat_id={chat_id}, 消息数={len(messages)}, hasMore={has_more}, nextCursor={next_cursor}")
                    return response
                else:
                    logger.error(f"[IM] ❌ 获取历史消息失败: code={code}")
                    logger.error(f"[IM] 完整响应: {response}")
                    return None
                    
            except asyncio.TimeoutError:
                logger.error(f"[IM] ❌ 获取历史消息超时: chat_id={chat_id}")
                # 清理pending请求
                self._pending_requests.pop(mid, None)
                return None
                
        except Exception as e:
            logger.error(f"[IM] ❌ 获取历史消息异常: {e}", exc_info=True)
            return None
    
    async def sync_chat_history(self, chat_id: str, limit: int = 100, conversation_seller_id: str = None) -> int:
        """
        同步指定会话的历史消息
        
        Args:
            chat_id: 会话ID
            limit: 最多同步多少条消息（默认100）
            conversation_seller_id: 会话的seller_id（用于补充发送消息的接收者信息）
        
        Returns:
            同步的消息数量
        """
        logger.debug(f"[IM] 📥 开始同步会话: chat_id={chat_id}, conversation_seller_id={conversation_seller_id}")
        
        cursor = 9007199254740991  # 从最新开始
        synced_count = 0
        
        try:
            while synced_count < limit:
                # 获取一页消息
                logger.info(f"[IM] 🔍 请求历史消息: chat_id={chat_id}, cursor={cursor}")
                response = await self.list_user_messages(chat_id, cursor, page_size=20)
                
                # 🔍 打印完整响应用于调试
                logger.info(f"[IM] 📥 API响应: code={response.get('code') if response else None}")
                if response:
                    body = response.get('body', {})
                    logger.info(f"[IM] 📥 响应body keys: {list(body.keys()) if body else None}")
                    if body:
                        messages = body.get('userMessageModels', [])
                        logger.info(f"[IM] 📥 消息数量: {len(messages) if messages else 0}")
                
                if not response or response.get('code') != 200:
                    logger.warning(f"[IM] 获取历史消息失败，停止同步: chat_id={chat_id}")
                    logger.warning(f"[IM] 完整响应: {response}")
                    break
                
                body = response.get('body', {})
                messages = body.get('userMessageModels', [])
                next_cursor = body.get('nextCursor')
                has_more = body.get('hasMore', False)
                
                # ✅ 批量处理优化：先解析所有消息，再批量发送
                logger.info(f"[IM] 🔄 开始批量解析 {len(messages)} 条消息")
                batch_messages = []
                for i, msg_model in enumerate(messages):
                    try:
                        logger.debug(f"[IM] 解析消息 {i+1}/{len(messages)}: message_id={msg_model.get('message', {}).get('messageId')}")
                        # ✅ 只解析，不立即发送
                        callback_data = await self._parse_history_message(msg_model, chat_id, conversation_seller_id)
                        if callback_data:
                            batch_messages.append(callback_data)
                        synced_count += 1
                        
                        # 达到限制
                        if synced_count >= limit:
                            break
                    except Exception as e:
                        logger.error(f"[IM] 解析历史消息失败: {e}", exc_info=True)
                        continue
                
                # ✅ 批量发送到server（一个TCP消息，server端用一个数据库连接处理）
                if batch_messages:
                    logger.info(f"[IM] 📦 批量发送 {len(batch_messages)} 条消息到server")
                    await self._send_batch_history_messages(batch_messages)
                
                # 检查是否还有更多
                if not has_more or not next_cursor or synced_count >= limit:
                    break
                
                cursor = next_cursor
                
                # 避免请求过快
                await asyncio.sleep(0.5)
            
            logger.debug(f"[IM] ✅ 会话 {chat_id} 同步完成: {synced_count}条消息")
            return synced_count
            
        except Exception as e:
            logger.error(f"[IM] ❌ 同步会话历史失败: chat_id={chat_id}, {e}", exc_info=True)
            return synced_count
    
    async def _parse_history_message(self, msg_model: dict, chat_id: str, conversation_seller_id: str = None) -> dict:
        """
        解析历史消息（仅解析，不发送）
        
        Args:
            msg_model: userMessageModel对象
            chat_id: 会话ID
            conversation_seller_id: 会话的seller_id（用于补充发送消息的接收者信息）
        
        Returns:
            callback_data: 解析后的消息数据，如果解析失败返回None
        """
        try:
            message = msg_model.get('message', {})
            
            # 基础字段
            message_id = message.get('messageId')
            create_at = message.get('createAt')  # 毫秒时间戳
            
            if not message_id:
                logger.warning(f"[IM] 历史消息缺少message_id，跳过")
                return
            
            # 解析消息内容
            content = message.get('content', {})
            custom = content.get('custom', {})
            
            # 获取消息类型
            custom_type = custom.get('type')  # 1=文本,2=图片,3=语音,4=视频,6=系统
            
            # 解析custom.data (base64编码)
            data_b64 = custom.get('data', '')
            if not data_b64:
                logger.warning(f"[IM] 历史消息缺少custom.data: message_id={message_id}")
                return
            
            # base64解码
            try:
                data_json_str = base64.b64decode(data_b64).decode('utf-8')
                data_json = json.loads(data_json_str)
            except Exception as e:
                logger.error(f"[IM] 解码custom.data失败: {e}")
                return
            
            # 根据contentType解析内容
            content_type_code = data_json.get('contentType')
            
            # 初始化字段
            message_type = 'text'
            custom_content_type = 1
            content_type_category = 'private'
            message_content = custom.get('summary', '')
            message_extra_dict = {}
            
            # 解析不同类型的消息
            if content_type_code == 1:  # 文本
                message_type = 'text'
                custom_content_type = 1
                text_obj = data_json.get('text', {})
                message_content = text_obj.get('text', '')
                # ✅ 如果text字段为空，尝试从summary获取
                if not message_content:
                    message_content = custom.get('summary', '')
                logger.debug(f"[IM] 历史文本消息: content={message_content[:50] if message_content else 'EMPTY'}")
                
            elif content_type_code == 2:  # 图片
                message_type = 'image'
                custom_content_type = 2
                image_obj = data_json.get('image', {})
                pics = image_obj.get('pics', [])
                if pics and len(pics) > 0:
                    pic = pics[0]
                    message_extra_dict['image_url'] = pic.get('url')
                    message_extra_dict['image_width'] = pic.get('width')
                    message_extra_dict['image_height'] = pic.get('height')
                    message_content = '[图片]'
                    logger.debug(f"[IM] 历史图片消息: url={pic.get('url')}, size={pic.get('width')}x{pic.get('height')}")
                
            elif content_type_code == 3:  # 语音
                message_type = 'voice'
                custom_content_type = 3
                audio_obj = data_json.get('audio', {})
                message_extra_dict['voice_url'] = audio_obj.get('url')
                message_extra_dict['voice_duration'] = audio_obj.get('duration')
                message_extra_dict['voice_size_bytes'] = audio_obj.get('sizeBytes')
                message_content = '[语音]'
                logger.debug(f"[IM] 历史语音消息: url={audio_obj.get('url')}, duration={audio_obj.get('duration')}s")
                
            elif content_type_code == 4:  # 视频
                message_type = 'video'
                custom_content_type = 4
                video_obj = data_json.get('video', {})
                message_extra_dict['video_url'] = video_obj.get('url')
                message_extra_dict['video_cover'] = video_obj.get('cover')
                message_extra_dict['video_duration'] = video_obj.get('duration')
                message_content = '[视频]'
                
            elif content_type_code == 6:  # 系统消息
                message_type = 'system'
                custom_content_type = 6
                content_type_category = 'system'
                text_card = data_json.get('textCard', {})
                if text_card:
                    message_extra_dict['system_data'] = {
                        'contentType': 6,
                        'title': text_card.get('title', ''),
                        'content': text_card.get('content', ''),
                        'labelIcon': text_card.get('labelIcon', ''),
                        'raw_textCard': text_card
                    }
                    message_content = text_card.get('title', '') or text_card.get('content', '')
            
            # 判断消息方向（发送者）
            sender = message.get('sender', {})
            sender_uid = sender.get('uid', '').replace('@goofish', '')
            sender_nick = sender.get('nick', '')
            
            # 如果sender_uid等于myid，说明是我们发送的
            direction = 'send' if str(sender_uid) == str(self.myid) else 'recv'
            
            # 提取item_id（如果有）
            item_id = None
            ext = message.get('ext', '')
            if ext:
                try:
                    ext_json = json.loads(ext)
                    item_id = ext_json.get('itemId')
                except:
                    pass
            
            # 构建回调数据
            message_extra_json = json.dumps(message_extra_dict, ensure_ascii=False) if message_extra_dict else None
            
            callback_data = {
                'chat_id': chat_id,
                'message_id': message_id,
                'message_type': message_type,
                'custom_content_type': custom_content_type,
                'content_type': content_type_category,
                'direction': direction,
                'message_content': message_content,
                'message_extra': message_extra_json,
                'message_time': self._parse_timestamp(create_at),  # ✅ 修复：使用message_time替代create_time
                'sender_id': sender_uid,
                'sender_name': sender_nick,
                'item_id': item_id,
                'client_id': self.state.get('client_id'),
                'account_user_id': str(self.myid),
                'is_history_sync': True,  # 标记为历史同步消息
            }
            
            # ✅ 如果是图片消息，添加图片信息到独立字段
            if message_type == 'image' and message_extra_dict:
                callback_data['image_url'] = message_extra_dict.get('image_url')
                callback_data['image_width'] = message_extra_dict.get('image_width')
                callback_data['image_height'] = message_extra_dict.get('image_height')
            
            # ✅ 如果是语音消息，添加语音信息到独立字段
            if message_type == 'voice' and message_extra_dict:
                callback_data['voice_url'] = message_extra_dict.get('voice_url')
                callback_data['voice_duration'] = message_extra_dict.get('voice_duration')
                callback_data['voice_size_bytes'] = message_extra_dict.get('voice_size_bytes')
            
            # ✅ 如果是视频消息，添加视频信息到独立字段
            if message_type == 'video' and message_extra_dict:
                callback_data['video_url'] = message_extra_dict.get('video_url')
                callback_data['video_cover'] = message_extra_dict.get('video_cover')
                callback_data['video_duration'] = message_extra_dict.get('video_duration')
            
            # ✅ 如果是系统消息，添加系统消息数据到独立字段
            if message_type == 'system' and message_extra_dict:
                callback_data['system_data'] = message_extra_dict.get('system_data')
            
            # 如果是接收的消息，seller_id是发送者
            # 如果是发送的消息，seller_id需要从会话中获取对方ID
            if direction == 'recv':
                callback_data['seller_id'] = sender_uid
                callback_data['seller_name'] = sender_nick
                logger.debug(f"[IM] 接收消息: seller_id={sender_uid}, seller_name={sender_nick}")
            else:
                # 发送的消息，需要从message.actualReceivers获取接收者
                actual_receivers = message.get('actualReceivers', [])
                if actual_receivers and len(actual_receivers) > 0:
                    receiver = actual_receivers[0]
                    receiver_uid = receiver.get('uid', '').replace('@goofish', '')
                    receiver_nick = receiver.get('nick', '')
                    callback_data['seller_id'] = receiver_uid
                    callback_data['seller_name'] = receiver_nick
                    logger.debug(f"[IM] 发送消息: seller_id={receiver_uid}, seller_name={receiver_nick} (from actualReceivers)")
                else:
                    # ⚠️ 如果没有actualReceivers，使用会话的seller_id作为后备
                    if conversation_seller_id:
                        callback_data['seller_id'] = conversation_seller_id
                        callback_data['seller_name'] = ''
                        logger.info(f"[IM] ✅ 发送消息缺少actualReceivers，使用会话seller_id: {conversation_seller_id}, message_id={message_id}")
                    else:
                        logger.warning(f"[IM] ⚠️ 发送的消息缺少actualReceivers且无会话seller_id: message_id={message_id}, sender={sender_uid}")
                        callback_data['seller_id'] = None
                        callback_data['seller_name'] = ''
            
            # ✅ 返回解析后的数据（不立即发送）
            logger.debug(f"[IM] ✅ 已解析历史消息: message_id={message_id}, type={message_type}, direction={direction}, seller_id={callback_data.get('seller_id')}")
            return callback_data
            
        except Exception as e:
            logger.error(f"[IM] ❌ 解析历史消息失败: {e}", exc_info=True)
            logger.debug(f"[IM] 原始消息: {json.dumps(msg_model, ensure_ascii=False)[:500]}")
            return None
    
    async def _send_batch_history_messages(self, batch_messages: list):
        """
        批量发送历史消息到server（一个TCP消息，server端用一个数据库连接处理）
        
        Args:
            batch_messages: 消息数组，每个元素是callback_data
        """
        if not self.message_callback or not batch_messages:
            return
        
        try:
            # ✅ 打印前3条消息用于调试
            if not hasattr(self, '_history_batch_debug_count'):
                self._history_batch_debug_count = 0
            if self._history_batch_debug_count < 3 and batch_messages:
                logger.info(f"[IM] 🔍 批量消息样本 (第一条): {json.dumps(batch_messages[0], ensure_ascii=False, indent=2)[:500]}")
                self._history_batch_debug_count += 1
            
            # ✅ 批量发送（通过特殊事件发送）
            batch_data = {
                'event': 'batch_history_messages',  # 特殊事件类型
                'direction': 'res',  # 响应方向
                'messages': batch_messages,
                'count': len(batch_messages),
                'is_history_sync': True
            }
            
            await self.message_callback(batch_data)
            logger.info(f"[IM] ✅ 已批量发送 {len(batch_messages)} 条历史消息到server")
            
        except Exception as e:
            logger.error(f"[IM] ❌ 批量发送历史消息失败: {e}", exc_info=True)
    
    async def sync_all_conversations_history(self, limit_per_conv: int = 50):
        """
        同步所有会话的历史消息
        
        Args:
            limit_per_conv: 每个会话最多同步多少条消息
        """
        logger.info("[IM] 📥 开始同步所有会话的历史消息...")
        
        try:
            # 获取会话列表（返回的是body对象，不是完整response）
            body = await self.get_conversation_list()
            
            if not body:
                logger.error("[IM] ❌ 获取会话列表失败，无法同步历史消息")
                return
            
            # get_conversation_list返回的body中，会话列表在userConvs字段
            conversations = body.get('userConvs', [])
            
            if not conversations:
                logger.info("[IM] 没有会话需要同步")
                return
            
            logger.info(f"[IM] 发现{len(conversations)}个会话，开始同步...")
            
            # 🔍 打印第一个会话的完整结构用于调试
            if conversations:
                first_conv = conversations[0]
                logger.info(f"[IM] 🔍 第一个会话的keys: {list(first_conv.keys())}")
                logger.info(f"[IM] 🔍 第一个会话的完整数据: {first_conv}")
            
            total_synced = 0
            skipped_empty = 0
            skipped_group = 0
            processed_count = 0  # 实际处理的会话数
            
            for conv in conversations:
                try:
                    # 只处理单聊（注意：字段名是type，不是conversationType）
                    conv_type = conv.get('type')
                    
                    if conv_type != 1:  # 1=单聊
                        skipped_group += 1
                        continue
                    
                    # ✅ 修复：cid在嵌套结构中，不是顶层
                    # 路径：singleChatUserConversation -> singleChatConversation -> cid
                    single_chat = conv.get('singleChatUserConversation', {})
                    conversation = single_chat.get('singleChatConversation', {})
                    cid = conversation.get('cid', '')
                    
                    logger.info(f"[IM] 🔍 检查会话: cid={cid}, type={conv_type}")
                    
                    # 提取chat_id（去掉@goofish后缀）
                    chat_id = cid.replace('@goofish', '')
                    
                    if not chat_id:
                        logger.warning(f"[IM] ⚠️ 会话缺少cid，跳过")
                        continue
                    
                    # ⚠️ 注意：会话列表中的lastMessage可能不存在，但不代表会话没有历史消息
                    # 历史消息需要通过list_user_messages API获取，不依赖lastMessage字段
                    last_message = conv.get('lastMessage')
                    if not last_message:
                        logger.info(f"[IM] 会话无lastMessage字段，尝试同步历史: chat_id={chat_id}")
                        # 不跳过，继续尝试获取历史消息
                    
                    # ✅ 提取会话的seller_id（用于补充发送消息的接收者信息）
                    extension = conversation.get("extension", {})
                    conversation_seller_id = extension.get("itemSellerId") or extension.get("ownerUserId") or extension.get("extUserId")
                    
                    # 同步这个会话的历史
                    processed_count += 1
                    logger.info(f"[IM] 📥 开始同步会话 {processed_count}/{len(conversations)}: chat_id={chat_id}, seller_id={conversation_seller_id}")
                    
                    count = await self.sync_chat_history(chat_id, limit=limit_per_conv, conversation_seller_id=conversation_seller_id)
                    
                    if count == 0:
                        skipped_empty += 1  # 确实没有消息
                        logger.info(f"[IM] ❌ 会话 {chat_id} 没有同步到消息")
                    else:
                        logger.info(f"[IM] ✅ 会话 {chat_id} 同步了 {count} 条消息")
                    
                    total_synced += count
                    
                    # 避免请求过快
                    await asyncio.sleep(1)
                    
                except Exception as e:
                    logger.error(f"[IM] 同步单个会话失败: {e}")
                    continue
            
            # 统计信息
            logger.info(f"[IM] ✅ 所有会话历史同步完成: 共同步{total_synced}条消息 | 处理{processed_count}个会话 (跳过: 群聊{skipped_group}个, 空会话{skipped_empty}个)")
            
        except Exception as e:
            logger.error(f"[IM] ❌ 同步所有会话历史失败: {e}", exc_info=True)

